You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by ke...@apache.org on 2022/08/20 07:25:02 UTC
[skywalking] branch master updated: Add Kubernetes support module to share codes between modules and reduce calls to Kubernetes API server (#9493)
This is an automated email from the ASF dual-hosted git repository.
kezhenxu94 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/skywalking.git
The following commit(s) were added to refs/heads/master by this push:
new e3d091ebd5 Add Kubernetes support module to share codes between modules and reduce calls to Kubernetes API server (#9493)
e3d091ebd5 is described below
commit e3d091ebd55d673db0214b4833a6dd147b5f8ef1
Author: kezhenxu94 <ke...@apache.org>
AuthorDate: Sat Aug 20 15:24:44 2022 +0800
Add Kubernetes support module to share codes between modules and reduce calls to Kubernetes API server (#9493)
---
docs/en/changes/changes.md | 1 +
oap-server/analyzer/meter-analyzer/pom.xml | 9 +-
.../oap/meter/analyzer/k8s/K8sInfoRegistry.java | 161 +++--------
.../oap/meter/analyzer/dsl/K8sTagTest.java | 18 +-
.../{ => library-kubernetes-support}/pom.xml | 20 +-
.../library/kubernetes/KubernetesClient.java | 40 +++
.../kubernetes/KubernetesEndpointWatcher.java | 107 +++++++
.../kubernetes/KubernetesEndpointsListener.java | 30 ++
.../library/kubernetes/KubernetesNodeListener.java | 30 ++
.../library/kubernetes/KubernetesNodeWatcher.java | 106 +++++++
.../library/kubernetes/KubernetesPodListener.java | 30 ++
.../library/kubernetes/KubernetesPodWatcher.java | 106 +++++++
.../kubernetes/KubernetesServiceListener.java | 30 ++
.../kubernetes/KubernetesServiceWatcher.java | 106 +++++++
oap-server/server-library/pom.xml | 1 +
.../envoy-metrics-receiver-plugin/pom.xml | 5 +-
.../receiver/envoy/als/k8s/K8SServiceRegistry.java | 307 ++++++++-------------
.../envoy/als/k8s/KubernetesNodeRegistry.java | 129 ---------
18 files changed, 767 insertions(+), 469 deletions(-)
diff --git a/docs/en/changes/changes.md b/docs/en/changes/changes.md
index 323126e1f8..0fcfcac382 100644
--- a/docs/en/changes/changes.md
+++ b/docs/en/changes/changes.md
@@ -52,6 +52,7 @@
* Support sending alarm messages to Discord.
* Fix query history process data failure.
* Optimize TTL mechanism for Elasticsearch storage, skip executed indices in one TTL rotation.
+* Add Kubernetes support module to share codes between modules and reduce calls to Kubernetes API server.
#### UI
diff --git a/oap-server/analyzer/meter-analyzer/pom.xml b/oap-server/analyzer/meter-analyzer/pom.xml
index 49dd2c1b77..3ef6dbb42f 100644
--- a/oap-server/analyzer/meter-analyzer/pom.xml
+++ b/oap-server/analyzer/meter-analyzer/pom.xml
@@ -28,6 +28,11 @@
<artifactId>meter-analyzer</artifactId>
<dependencies>
+ <dependency>
+ <groupId>org.apache.skywalking</groupId>
+ <artifactId>library-kubernetes-support</artifactId>
+ <version>${project.version}</version>
+ </dependency>
<dependency>
<groupId>org.apache.skywalking</groupId>
<artifactId>server-core</artifactId>
@@ -41,9 +46,5 @@
<groupId>io.vavr</groupId>
<artifactId>vavr</artifactId>
</dependency>
- <dependency>
- <groupId>io.kubernetes</groupId>
- <artifactId>client-java</artifactId>
- </dependency>
</dependencies>
</project>
diff --git a/oap-server/analyzer/meter-analyzer/src/main/java/org/apache/skywalking/oap/meter/analyzer/k8s/K8sInfoRegistry.java b/oap-server/analyzer/meter-analyzer/src/main/java/org/apache/skywalking/oap/meter/analyzer/k8s/K8sInfoRegistry.java
index e5efcfb0e6..5d3911824d 100644
--- a/oap-server/analyzer/meter-analyzer/src/main/java/org/apache/skywalking/oap/meter/analyzer/k8s/K8sInfoRegistry.java
+++ b/oap-server/analyzer/meter-analyzer/src/main/java/org/apache/skywalking/oap/meter/analyzer/k8s/K8sInfoRegistry.java
@@ -18,161 +18,58 @@
package org.apache.skywalking.oap.meter.analyzer.k8s;
+import static java.util.Objects.isNull;
+import static java.util.Optional.ofNullable;
+import java.util.Collection;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
+import org.apache.skywalking.library.kubernetes.KubernetesPodListener;
+import org.apache.skywalking.library.kubernetes.KubernetesPodWatcher;
+import org.apache.skywalking.library.kubernetes.KubernetesServiceListener;
+import org.apache.skywalking.library.kubernetes.KubernetesServiceWatcher;
+import org.apache.skywalking.oap.server.library.util.CollectionUtils;
+import org.apache.skywalking.oap.server.library.util.StringUtil;
import com.google.common.base.Strings;
-import com.google.common.util.concurrent.ThreadFactoryBuilder;
-import io.kubernetes.client.informer.ResourceEventHandler;
-import io.kubernetes.client.informer.SharedInformerFactory;
-import io.kubernetes.client.openapi.Configuration;
-import io.kubernetes.client.openapi.apis.CoreV1Api;
import io.kubernetes.client.openapi.models.V1LoadBalancerIngress;
import io.kubernetes.client.openapi.models.V1LoadBalancerStatus;
-import io.kubernetes.client.openapi.models.V1PodList;
+import io.kubernetes.client.openapi.models.V1Pod;
import io.kubernetes.client.openapi.models.V1Service;
-import io.kubernetes.client.openapi.models.V1ServiceList;
import io.kubernetes.client.openapi.models.V1ServiceSpec;
import io.kubernetes.client.openapi.models.V1ServiceStatus;
-import io.kubernetes.client.util.Config;
-import io.kubernetes.client.openapi.ApiClient;
-import io.kubernetes.client.openapi.models.V1Pod;
-import java.util.Collection;
-import java.util.Map;
-import java.util.Objects;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
import lombok.SneakyThrows;
-import lombok.extern.slf4j.Slf4j;
-import org.apache.skywalking.oap.server.library.util.CollectionUtils;
-import org.apache.skywalking.oap.server.library.util.StringUtil;
-import static java.util.Objects.isNull;
-import static java.util.Optional.ofNullable;
-
-@Slf4j
-public class K8sInfoRegistry {
+public class K8sInfoRegistry
+ implements KubernetesServiceListener, KubernetesPodListener {
private final static K8sInfoRegistry INSTANCE = new K8sInfoRegistry();
- private final AtomicBoolean isStarted = new AtomicBoolean(false);
private final Map<String/* podName.namespace */, V1Pod> namePodMap = new ConcurrentHashMap<>();
protected final Map<String/* serviceName.namespace */, V1Service> nameServiceMap = new ConcurrentHashMap<>();
private final Map<String/* podName.namespace */, String /* serviceName.namespace */> podServiceMap = new ConcurrentHashMap<>();
private final Map<String/* podIP */, String /* podName.namespace */> ipPodMap = new ConcurrentHashMap<>();
private final Map<String/* serviceIP */, String /* serviceName.namespace */> ipServiceMap = new ConcurrentHashMap<>();
- private ExecutorService executor;
private static final String SEPARATOR = ".";
public static K8sInfoRegistry getInstance() {
return INSTANCE;
}
- private void init() {
- executor = Executors.newCachedThreadPool(
- new ThreadFactoryBuilder()
- .setNameFormat("K8sInfoRegistry-%d")
- .setDaemon(true)
- .build()
- );
- }
-
@SneakyThrows
public void start() {
- if (isStarted.compareAndSet(false, true)) {
- init();
- final ApiClient apiClient = Config.defaultClient();
- apiClient.setHttpClient(apiClient.getHttpClient()
- .newBuilder()
- .readTimeout(0, TimeUnit.SECONDS)
- .build());
- Configuration.setDefaultApiClient(apiClient);
-
- final CoreV1Api coreV1Api = new CoreV1Api();
- final SharedInformerFactory factory = new SharedInformerFactory(executor);
- listenServiceEvents(coreV1Api, factory);
- listenPodEvents(coreV1Api, factory);
- factory.startAllRegisteredInformers();
- }
- }
-
- private void listenServiceEvents(final CoreV1Api coreV1Api, final SharedInformerFactory factory) {
- factory.sharedIndexInformerFor(
- params -> coreV1Api.listServiceForAllNamespacesCall(
- null,
- null,
- null,
- null,
- null,
- null,
- params.resourceVersion,
- null,
- params.timeoutSeconds,
- params.watch,
- null
- ),
- V1Service.class,
- V1ServiceList.class
- ).addEventHandler(new ResourceEventHandler<V1Service>() {
- @Override
- public void onAdd(final V1Service service) {
- addService(service);
- }
-
- @Override
- public void onUpdate(final V1Service oldService, final V1Service newService) {
- addService(newService);
- }
-
- @Override
- public void onDelete(final V1Service service, final boolean deletedFinalStateUnknown) {
- removeService(service);
- }
- });
- }
-
- private void listenPodEvents(final CoreV1Api coreV1Api, final SharedInformerFactory factory) {
- factory.sharedIndexInformerFor(
- params -> coreV1Api.listPodForAllNamespacesCall(
- null,
- null,
- null,
- null,
- null,
- null,
- params.resourceVersion,
- null,
- params.timeoutSeconds,
- params.watch,
- null
- ),
- V1Pod.class,
- V1PodList.class
- ).addEventHandler(new ResourceEventHandler<V1Pod>() {
- @Override
- public void onAdd(final V1Pod pod) {
- addPod(pod);
- }
-
- @Override
- public void onUpdate(final V1Pod oldPod, final V1Pod newPod) {
- addPod(newPod);
- }
-
- @Override
- public void onDelete(final V1Pod pod, final boolean deletedFinalStateUnknown) {
- removePod(pod);
- }
- });
+ KubernetesPodWatcher.INSTANCE.addListener(this).start();
+ KubernetesServiceWatcher.INSTANCE.addListener(this).start();
}
- protected void addService(final V1Service service) {
+ @Override
+ public void onServiceAdded(final V1Service service) {
ofNullable(service.getMetadata()).ifPresent(
metadata -> nameServiceMap.put(metadata.getName() + SEPARATOR + metadata.getNamespace(), service)
);
recompose();
}
- protected void removeService(final V1Service service) {
+ @Override
+ public void onServiceDeleted(final V1Service service) {
ofNullable(service.getMetadata()).ifPresent(
metadata -> nameServiceMap.remove(metadata.getName() + SEPARATOR + metadata.getNamespace())
);
@@ -185,14 +82,21 @@ public class K8sInfoRegistry {
recompose();
}
- protected void addPod(final V1Pod pod) {
+ @Override
+ public void onServiceUpdated(V1Service oldService, V1Service newService) {
+ onServiceAdded(newService);
+ }
+
+ @Override
+ public void onPodAdded(final V1Pod pod) {
ofNullable(pod.getMetadata()).ifPresent(
metadata -> namePodMap.put(metadata.getName() + SEPARATOR + metadata.getNamespace(), pod));
recompose();
}
- protected void removePod(final V1Pod pod) {
+ @Override
+ public void onPodDeleted(final V1Pod pod) {
ofNullable(pod.getMetadata()).ifPresent(
metadata -> namePodMap.remove(metadata.getName() + SEPARATOR + metadata.getNamespace()));
@@ -203,6 +107,11 @@ public class K8sInfoRegistry {
status -> ipPodMap.remove(status.getPodIP()));
}
+ @Override
+ public void onPodUpdated(V1Pod oldPod, V1Pod newPod) {
+ onPodAdded(newPod);
+ }
+
private void recompose() {
namePodMap.forEach((podName, pod) -> {
if (!isNull(pod.getMetadata())) {
diff --git a/oap-server/analyzer/meter-analyzer/src/test/java/org/apache/skywalking/oap/meter/analyzer/dsl/K8sTagTest.java b/oap-server/analyzer/meter-analyzer/src/test/java/org/apache/skywalking/oap/meter/analyzer/dsl/K8sTagTest.java
index caa39ac8d0..9c9ce5867a 100644
--- a/oap-server/analyzer/meter-analyzer/src/test/java/org/apache/skywalking/oap/meter/analyzer/dsl/K8sTagTest.java
+++ b/oap-server/analyzer/meter-analyzer/src/test/java/org/apache/skywalking/oap/meter/analyzer/dsl/K8sTagTest.java
@@ -253,33 +253,33 @@ public class K8sTagTest {
PowerMockito.doNothing().when(K8sInfoRegistry.getInstance(), "start");
PowerMockito.when(
- K8sInfoRegistry.getInstance(), "addService", mockService("nginx-service", "default", of("run", "nginx"), "2.2.2.1"))
+ K8sInfoRegistry.getInstance(), "onServiceAdded", mockService("nginx-service", "default", of("run", "nginx"), "2.2.2.1"))
.thenCallRealMethod();
PowerMockito.when(
- K8sInfoRegistry.getInstance(), "addService",
+ K8sInfoRegistry.getInstance(), "onServiceAdded",
mockService("kube-state-metrics", "kube-system", of("run", "kube-state-metrics"), "2.2.2.2")
).thenCallRealMethod();
PowerMockito.when(
- K8sInfoRegistry.getInstance(), "addPod",
+ K8sInfoRegistry.getInstance(), "onPodAdded",
mockPod("my-nginx-5dc4865748-mbczh", "default", of("run", "nginx"), "1.1.1.1")
).thenCallRealMethod();
PowerMockito.when(
- K8sInfoRegistry.getInstance(), "addPod",
+ K8sInfoRegistry.getInstance(), "onPodAdded",
mockPod("kube-state-metrics-6f979fd498-z7xwx", "kube-system", of("run", "kube-state-metrics"), "1.1.1.2")
).thenCallRealMethod();
PowerMockito.when(
- K8sInfoRegistry.getInstance(), "removeService", mockService("nginx-service", "default", of("run", "nginx"), "2.2.2.1"))
+ K8sInfoRegistry.getInstance(), "onServiceDeleted", mockService("nginx-service", "default", of("run", "nginx"), "2.2.2.1"))
.thenCallRealMethod();
PowerMockito.when(
- K8sInfoRegistry.getInstance(), "removePod",
+ K8sInfoRegistry.getInstance(), "onPodDeleted",
mockPod("my-nginx-5dc4865748-mbczh", "default", of("run", "nginx"), "1.1.1.1")
).thenCallRealMethod();
PowerMockito.when(
- K8sInfoRegistry.getInstance(), "addService", mockService("nginx-service", "default", of("run", "nginx"), "2.2.2.1"))
+ K8sInfoRegistry.getInstance(), "onServiceAdded", mockService("nginx-service", "default", of("run", "nginx"), "2.2.2.1"))
.thenCallRealMethod();
PowerMockito.when(
- K8sInfoRegistry.getInstance(), "addPod",
+ K8sInfoRegistry.getInstance(), "onPodAdded",
mockPod("my-nginx-5dc4865748-mbczh", "default", of("run", "nginx"), "1.1.1.1")
).thenCallRealMethod();
@@ -339,4 +339,4 @@ public class K8sTagTest {
return v1Pod;
}
-}
\ No newline at end of file
+}
diff --git a/oap-server/server-library/pom.xml b/oap-server/server-library/library-kubernetes-support/pom.xml
similarity index 75%
copy from oap-server/server-library/pom.xml
copy to oap-server/server-library/library-kubernetes-support/pom.xml
index 528ecc93b8..74387bf40e 100644
--- a/oap-server/server-library/pom.xml
+++ b/oap-server/server-library/library-kubernetes-support/pom.xml
@@ -19,20 +19,18 @@
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
- <artifactId>oap-server</artifactId>
+ <artifactId>server-library</artifactId>
<groupId>org.apache.skywalking</groupId>
<version>9.2.0-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
- <artifactId>server-library</artifactId>
- <packaging>pom</packaging>
- <modules>
- <module>library-module</module>
- <module>library-server</module>
- <module>library-util</module>
- <module>library-client</module>
- <module>library-elasticsearch-client</module>
- <module>library-datacarrier-queue</module>
- </modules>
+ <artifactId>library-kubernetes-support</artifactId>
+
+ <dependencies>
+ <dependency>
+ <groupId>io.kubernetes</groupId>
+ <artifactId>client-java</artifactId>
+ </dependency>
+ </dependencies>
</project>
diff --git a/oap-server/server-library/library-kubernetes-support/src/main/java/org/apache/skywalking/library/kubernetes/KubernetesClient.java b/oap-server/server-library/library-kubernetes-support/src/main/java/org/apache/skywalking/library/kubernetes/KubernetesClient.java
new file mode 100644
index 0000000000..825751a9ef
--- /dev/null
+++ b/oap-server/server-library/library-kubernetes-support/src/main/java/org/apache/skywalking/library/kubernetes/KubernetesClient.java
@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.skywalking.library.kubernetes;
+
+import java.io.IOException;
+import java.util.concurrent.TimeUnit;
+import io.kubernetes.client.openapi.ApiClient;
+import io.kubernetes.client.openapi.Configuration;
+import io.kubernetes.client.util.Config;
+import lombok.experimental.UtilityClass;
+
+@UtilityClass
+public class KubernetesClient {
+ public void setDefault() throws IOException {
+ final ApiClient apiClient = Config.defaultClient();
+ apiClient
+ .setHttpClient(apiClient.getHttpClient()
+ .newBuilder()
+ .readTimeout(0, TimeUnit.SECONDS)
+ .build());
+ Configuration.setDefaultApiClient(apiClient);
+ }
+}
diff --git a/oap-server/server-library/library-kubernetes-support/src/main/java/org/apache/skywalking/library/kubernetes/KubernetesEndpointWatcher.java b/oap-server/server-library/library-kubernetes-support/src/main/java/org/apache/skywalking/library/kubernetes/KubernetesEndpointWatcher.java
new file mode 100644
index 0000000000..8fd311290b
--- /dev/null
+++ b/oap-server/server-library/library-kubernetes-support/src/main/java/org/apache/skywalking/library/kubernetes/KubernetesEndpointWatcher.java
@@ -0,0 +1,107 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.skywalking.library.kubernetes;
+
+import static java.util.Objects.requireNonNull;
+import java.io.IOException;
+import java.util.Collections;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicBoolean;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import io.kubernetes.client.informer.ResourceEventHandler;
+import io.kubernetes.client.informer.SharedInformerFactory;
+import io.kubernetes.client.openapi.apis.CoreV1Api;
+import io.kubernetes.client.openapi.models.V1Endpoints;
+import io.kubernetes.client.openapi.models.V1EndpointsList;
+
+public enum KubernetesEndpointWatcher implements ResourceEventHandler<V1Endpoints> {
+ INSTANCE;
+
+ private final ExecutorService executor = Executors.newCachedThreadPool(
+ new ThreadFactoryBuilder()
+ .setNameFormat("KubernetesEndpointWatcher-%d")
+ .setDaemon(true)
+ .build()
+ );
+ private final AtomicBoolean started = new AtomicBoolean();
+ private final Set<KubernetesEndpointsListener> listeners = Collections.newSetFromMap(new ConcurrentHashMap<>());
+
+ public void start() throws IOException {
+ if (!started.compareAndSet(false, true)) {
+ return;
+ }
+
+ KubernetesClient.setDefault();
+
+ final CoreV1Api coreV1Api = new CoreV1Api();
+ final SharedInformerFactory factory = new SharedInformerFactory(executor);
+
+ listenEndpointEvents(coreV1Api, factory);
+
+ factory.startAllRegisteredInformers();
+ }
+
+ public KubernetesEndpointWatcher addListener(KubernetesEndpointsListener listener) {
+ requireNonNull(listener, "listener");
+
+ listeners.add(listener);
+
+ return this;
+ }
+
+ @Override
+ public void onAdd(V1Endpoints endpoints) {
+ listeners.forEach(it -> it.onEndpointsAdded(endpoints));
+ }
+
+ @Override
+ public void onUpdate(V1Endpoints oldEndpoints, V1Endpoints newEndpoints) {
+ listeners.forEach(it -> it.onEndpointsUpdated(oldEndpoints, newEndpoints));
+ }
+
+ @Override
+ public void onDelete(V1Endpoints endpoints, boolean deletedFinalStateUnknown) {
+ listeners.forEach(it -> it.onEndpointsDeleted(endpoints));
+ }
+
+ private void listenEndpointEvents(final CoreV1Api coreV1Api,
+ final SharedInformerFactory factory) {
+ factory.sharedIndexInformerFor(
+ params -> coreV1Api.listEndpointsForAllNamespacesCall(
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ params.resourceVersion,
+ null,
+ params.timeoutSeconds,
+ params.watch,
+ null
+ ),
+ V1Endpoints.class,
+ V1EndpointsList.class
+ ).addEventHandler(this);
+ }
+}
diff --git a/oap-server/server-library/library-kubernetes-support/src/main/java/org/apache/skywalking/library/kubernetes/KubernetesEndpointsListener.java b/oap-server/server-library/library-kubernetes-support/src/main/java/org/apache/skywalking/library/kubernetes/KubernetesEndpointsListener.java
new file mode 100644
index 0000000000..960b790d5d
--- /dev/null
+++ b/oap-server/server-library/library-kubernetes-support/src/main/java/org/apache/skywalking/library/kubernetes/KubernetesEndpointsListener.java
@@ -0,0 +1,30 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.skywalking.library.kubernetes;
+
+import io.kubernetes.client.openapi.models.V1Endpoints;
+
+public interface KubernetesEndpointsListener {
+ public void onEndpointsAdded(final V1Endpoints endpoints);
+
+ public void onEndpointsUpdated(final V1Endpoints oldEndpoints, final V1Endpoints newEndpoints);
+
+ public void onEndpointsDeleted(final V1Endpoints endpoints);
+}
diff --git a/oap-server/server-library/library-kubernetes-support/src/main/java/org/apache/skywalking/library/kubernetes/KubernetesNodeListener.java b/oap-server/server-library/library-kubernetes-support/src/main/java/org/apache/skywalking/library/kubernetes/KubernetesNodeListener.java
new file mode 100644
index 0000000000..1b6ed4c9eb
--- /dev/null
+++ b/oap-server/server-library/library-kubernetes-support/src/main/java/org/apache/skywalking/library/kubernetes/KubernetesNodeListener.java
@@ -0,0 +1,30 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.skywalking.library.kubernetes;
+
+import io.kubernetes.client.openapi.models.V1Node;
+
+public interface KubernetesNodeListener {
+ public void onNodeAdded(final V1Node node);
+
+ public void onNodeUpdated(final V1Node oldNode, final V1Node newNode);
+
+ public void onNodeDeleted(final V1Node node);
+}
diff --git a/oap-server/server-library/library-kubernetes-support/src/main/java/org/apache/skywalking/library/kubernetes/KubernetesNodeWatcher.java b/oap-server/server-library/library-kubernetes-support/src/main/java/org/apache/skywalking/library/kubernetes/KubernetesNodeWatcher.java
new file mode 100644
index 0000000000..7f87d33636
--- /dev/null
+++ b/oap-server/server-library/library-kubernetes-support/src/main/java/org/apache/skywalking/library/kubernetes/KubernetesNodeWatcher.java
@@ -0,0 +1,106 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.skywalking.library.kubernetes;
+
+import static java.util.Objects.requireNonNull;
+import java.io.IOException;
+import java.util.Set;
+import java.util.concurrent.CopyOnWriteArraySet;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicBoolean;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import io.kubernetes.client.informer.ResourceEventHandler;
+import io.kubernetes.client.informer.SharedInformerFactory;
+import io.kubernetes.client.openapi.apis.CoreV1Api;
+import io.kubernetes.client.openapi.models.V1Node;
+import io.kubernetes.client.openapi.models.V1NodeList;
+
+public enum KubernetesNodeWatcher implements ResourceEventHandler<V1Node> {
+ INSTANCE;
+
+ private final ExecutorService executor = Executors.newCachedThreadPool(
+ new ThreadFactoryBuilder()
+ .setNameFormat("KubernetesNodeWatcher-%d")
+ .setDaemon(true)
+ .build()
+ );
+ private final AtomicBoolean started = new AtomicBoolean();
+ private final Set<KubernetesNodeListener> listeners = new CopyOnWriteArraySet<>();
+
+ public void start() throws IOException {
+ if (!started.compareAndSet(false, true)) {
+ return;
+ }
+
+ KubernetesClient.setDefault();
+
+ final CoreV1Api coreV1Api = new CoreV1Api();
+ final SharedInformerFactory factory = new SharedInformerFactory(executor);
+
+ listenNodeEvents(coreV1Api, factory);
+
+ factory.startAllRegisteredInformers();
+ }
+
+ public KubernetesNodeWatcher addListener(KubernetesNodeListener listener) {
+ requireNonNull(listener, "listener");
+
+ listeners.add(listener);
+
+ return this;
+ }
+
+ @Override
+ public void onAdd(V1Node node) {
+ listeners.forEach(it -> it.onNodeAdded(node));
+ }
+
+ @Override
+ public void onUpdate(V1Node oldNode, V1Node newNode) {
+ listeners.forEach(it -> it.onNodeUpdated(oldNode, newNode));
+ }
+
+ @Override
+ public void onDelete(V1Node node, boolean deletedFinalStateUnknown) {
+ listeners.forEach(it -> it.onNodeDeleted(node));
+ }
+
+ private void listenNodeEvents(final CoreV1Api coreV1Api,
+ final SharedInformerFactory factory) {
+ factory.sharedIndexInformerFor(
+ params -> coreV1Api.listNodeCall(
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ params.resourceVersion,
+ null,
+ params.timeoutSeconds,
+ params.watch,
+ null
+ ),
+ V1Node.class,
+ V1NodeList.class
+ ).addEventHandler(this);
+ }
+}
diff --git a/oap-server/server-library/library-kubernetes-support/src/main/java/org/apache/skywalking/library/kubernetes/KubernetesPodListener.java b/oap-server/server-library/library-kubernetes-support/src/main/java/org/apache/skywalking/library/kubernetes/KubernetesPodListener.java
new file mode 100644
index 0000000000..6b01eb4636
--- /dev/null
+++ b/oap-server/server-library/library-kubernetes-support/src/main/java/org/apache/skywalking/library/kubernetes/KubernetesPodListener.java
@@ -0,0 +1,30 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.skywalking.library.kubernetes;
+
+import io.kubernetes.client.openapi.models.V1Pod;
+
+public interface KubernetesPodListener {
+ public void onPodAdded(final V1Pod pod);
+
+ public void onPodUpdated(final V1Pod oldPod, final V1Pod newPod);
+
+ public void onPodDeleted(final V1Pod pod);
+}
diff --git a/oap-server/server-library/library-kubernetes-support/src/main/java/org/apache/skywalking/library/kubernetes/KubernetesPodWatcher.java b/oap-server/server-library/library-kubernetes-support/src/main/java/org/apache/skywalking/library/kubernetes/KubernetesPodWatcher.java
new file mode 100644
index 0000000000..04584788dc
--- /dev/null
+++ b/oap-server/server-library/library-kubernetes-support/src/main/java/org/apache/skywalking/library/kubernetes/KubernetesPodWatcher.java
@@ -0,0 +1,106 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.skywalking.library.kubernetes;
+
+import static java.util.Objects.requireNonNull;
+import java.io.IOException;
+import java.util.Set;
+import java.util.concurrent.CopyOnWriteArraySet;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicBoolean;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import io.kubernetes.client.informer.ResourceEventHandler;
+import io.kubernetes.client.informer.SharedInformerFactory;
+import io.kubernetes.client.openapi.apis.CoreV1Api;
+import io.kubernetes.client.openapi.models.V1Pod;
+import io.kubernetes.client.openapi.models.V1PodList;
+
+public enum KubernetesPodWatcher implements ResourceEventHandler<V1Pod> {
+ INSTANCE;
+
+ private final ExecutorService executor = Executors.newCachedThreadPool(
+ new ThreadFactoryBuilder()
+ .setNameFormat("KubernetesPodWatcher-%d")
+ .setDaemon(true)
+ .build()
+ );
+ private final AtomicBoolean started = new AtomicBoolean();
+ private final Set<KubernetesPodListener> listeners = new CopyOnWriteArraySet<>();
+
+ public void start() throws IOException {
+ if (!started.compareAndSet(false, true)) {
+ return;
+ }
+
+ KubernetesClient.setDefault();
+
+ final CoreV1Api coreV1Api = new CoreV1Api();
+ final SharedInformerFactory factory = new SharedInformerFactory(executor);
+
+ listenPodEvents(coreV1Api, factory);
+
+ factory.startAllRegisteredInformers();
+ }
+
+ public KubernetesPodWatcher addListener(KubernetesPodListener listener) {
+ requireNonNull(listener, "listener");
+
+ listeners.add(listener);
+
+ return this;
+ }
+
+ @Override
+ public void onAdd(V1Pod pod) {
+ listeners.forEach(it -> it.onPodAdded(pod));
+ }
+
+ @Override
+ public void onUpdate(V1Pod oldPod, V1Pod newPod) {
+ listeners.forEach(it -> it.onPodUpdated(oldPod, newPod));
+ }
+
+ @Override
+ public void onDelete(V1Pod pod, boolean deletedFinalStateUnknown) {
+ listeners.forEach(it -> it.onPodDeleted(pod));
+ }
+
+ private void listenPodEvents(final CoreV1Api coreV1Api,
+ final SharedInformerFactory factory) {
+ factory.sharedIndexInformerFor(
+ params -> coreV1Api.listPodForAllNamespacesCall(
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ params.resourceVersion,
+ null,
+ params.timeoutSeconds,
+ params.watch,
+ null
+ ),
+ V1Pod.class,
+ V1PodList.class
+ ).addEventHandler(this);
+ }
+}
diff --git a/oap-server/server-library/library-kubernetes-support/src/main/java/org/apache/skywalking/library/kubernetes/KubernetesServiceListener.java b/oap-server/server-library/library-kubernetes-support/src/main/java/org/apache/skywalking/library/kubernetes/KubernetesServiceListener.java
new file mode 100644
index 0000000000..21fe2a91d9
--- /dev/null
+++ b/oap-server/server-library/library-kubernetes-support/src/main/java/org/apache/skywalking/library/kubernetes/KubernetesServiceListener.java
@@ -0,0 +1,30 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.skywalking.library.kubernetes;
+
+import io.kubernetes.client.openapi.models.V1Service;
+
+public interface KubernetesServiceListener {
+ public void onServiceAdded(final V1Service service);
+
+ public void onServiceUpdated(final V1Service oldService, final V1Service newService);
+
+ public void onServiceDeleted(final V1Service service);
+}
diff --git a/oap-server/server-library/library-kubernetes-support/src/main/java/org/apache/skywalking/library/kubernetes/KubernetesServiceWatcher.java b/oap-server/server-library/library-kubernetes-support/src/main/java/org/apache/skywalking/library/kubernetes/KubernetesServiceWatcher.java
new file mode 100644
index 0000000000..00f1eaa75c
--- /dev/null
+++ b/oap-server/server-library/library-kubernetes-support/src/main/java/org/apache/skywalking/library/kubernetes/KubernetesServiceWatcher.java
@@ -0,0 +1,106 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.skywalking.library.kubernetes;
+
+import static java.util.Objects.requireNonNull;
+import java.io.IOException;
+import java.util.Set;
+import java.util.concurrent.CopyOnWriteArraySet;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicBoolean;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import io.kubernetes.client.informer.ResourceEventHandler;
+import io.kubernetes.client.informer.SharedInformerFactory;
+import io.kubernetes.client.openapi.apis.CoreV1Api;
+import io.kubernetes.client.openapi.models.V1Service;
+import io.kubernetes.client.openapi.models.V1ServiceList;
+
+public enum KubernetesServiceWatcher implements ResourceEventHandler<V1Service> {
+ INSTANCE;
+
+ private final ExecutorService executor = Executors.newCachedThreadPool(
+ new ThreadFactoryBuilder()
+ .setNameFormat("KubernetesServiceWatcher-%d")
+ .setDaemon(true)
+ .build()
+ );
+ private final AtomicBoolean started = new AtomicBoolean();
+ private final Set<KubernetesServiceListener> listeners = new CopyOnWriteArraySet<>();
+
+ public void start() throws IOException {
+ if (!started.compareAndSet(false, true)) {
+ return;
+ }
+
+ KubernetesClient.setDefault();
+
+ final CoreV1Api coreV1Api = new CoreV1Api();
+ final SharedInformerFactory factory = new SharedInformerFactory(executor);
+
+ listenServiceEvents(coreV1Api, factory);
+
+ factory.startAllRegisteredInformers();
+ }
+
+ public KubernetesServiceWatcher addListener(KubernetesServiceListener listener) {
+ requireNonNull(listener, "listener");
+
+ listeners.add(listener);
+
+ return this;
+ }
+
+ @Override
+ public void onAdd(V1Service service) {
+ listeners.forEach(it -> it.onServiceAdded(service));
+ }
+
+ @Override
+ public void onUpdate(V1Service oldService, V1Service newService) {
+ listeners.forEach(it -> it.onServiceUpdated(oldService, newService));
+ }
+
+ @Override
+ public void onDelete(V1Service service, boolean deletedFinalStateUnknown) {
+ listeners.forEach(it -> it.onServiceDeleted(service));
+ }
+
+ private void listenServiceEvents(final CoreV1Api coreV1Api,
+ final SharedInformerFactory factory) {
+ factory.sharedIndexInformerFor(
+ params -> coreV1Api.listServiceForAllNamespacesCall(
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ params.resourceVersion,
+ null,
+ params.timeoutSeconds,
+ params.watch,
+ null
+ ),
+ V1Service.class,
+ V1ServiceList.class
+ ).addEventHandler(this);
+ }
+}
diff --git a/oap-server/server-library/pom.xml b/oap-server/server-library/pom.xml
index 528ecc93b8..b58ca14a4d 100644
--- a/oap-server/server-library/pom.xml
+++ b/oap-server/server-library/pom.xml
@@ -34,5 +34,6 @@
<module>library-client</module>
<module>library-elasticsearch-client</module>
<module>library-datacarrier-queue</module>
+ <module>library-kubernetes-support</module>
</modules>
</project>
diff --git a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/pom.xml b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/pom.xml
index 9fc1fb26cb..c9f7aa6148 100644
--- a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/pom.xml
+++ b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/pom.xml
@@ -50,8 +50,9 @@
<version>${project.version}</version>
</dependency>
<dependency>
- <groupId>io.kubernetes</groupId>
- <artifactId>client-java</artifactId>
+ <groupId>org.apache.skywalking</groupId>
+ <artifactId>library-kubernetes-support</artifactId>
+ <version>${project.version}</version>
</dependency>
<dependency>
diff --git a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/k8s/K8SServiceRegistry.java b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/k8s/K8SServiceRegistry.java
index 14eaf1a17a..9902e147ce 100644
--- a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/k8s/K8SServiceRegistry.java
+++ b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/k8s/K8SServiceRegistry.java
@@ -18,40 +18,44 @@
package org.apache.skywalking.oap.server.receiver.envoy.als.k8s;
-import com.google.common.collect.ImmutableMap;
-import com.google.common.util.concurrent.ThreadFactoryBuilder;
-import io.kubernetes.client.informer.ResourceEventHandler;
-import io.kubernetes.client.informer.SharedInformerFactory;
-import io.kubernetes.client.openapi.ApiClient;
-import io.kubernetes.client.openapi.Configuration;
-import io.kubernetes.client.openapi.apis.CoreV1Api;
-import io.kubernetes.client.openapi.models.V1Endpoints;
-import io.kubernetes.client.openapi.models.V1EndpointsList;
-import io.kubernetes.client.openapi.models.V1ObjectMeta;
-import io.kubernetes.client.openapi.models.V1Pod;
-import io.kubernetes.client.openapi.models.V1PodList;
-import io.kubernetes.client.openapi.models.V1Service;
-import io.kubernetes.client.openapi.models.V1ServiceList;
-import io.kubernetes.client.util.Config;
+import static com.google.common.base.Strings.isNullOrEmpty;
+import static java.util.Objects.isNull;
+import static java.util.Optional.ofNullable;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
import java.util.stream.Collectors;
-import lombok.extern.slf4j.Slf4j;
+import org.apache.skywalking.library.kubernetes.KubernetesEndpointWatcher;
+import org.apache.skywalking.library.kubernetes.KubernetesEndpointsListener;
+import org.apache.skywalking.library.kubernetes.KubernetesNodeListener;
+import org.apache.skywalking.library.kubernetes.KubernetesNodeWatcher;
+import org.apache.skywalking.library.kubernetes.KubernetesPodListener;
+import org.apache.skywalking.library.kubernetes.KubernetesPodWatcher;
+import org.apache.skywalking.library.kubernetes.KubernetesServiceListener;
+import org.apache.skywalking.library.kubernetes.KubernetesServiceWatcher;
+import org.apache.skywalking.oap.server.library.util.StringUtil;
import org.apache.skywalking.oap.server.receiver.envoy.EnvoyMetricReceiverConfig;
import org.apache.skywalking.oap.server.receiver.envoy.als.ServiceMetaInfo;
-
-import static com.google.common.base.Strings.isNullOrEmpty;
-import static java.util.Objects.isNull;
-import static java.util.Optional.ofNullable;
+import com.google.common.collect.ImmutableMap;
+import io.kubernetes.client.openapi.models.V1Endpoints;
+import io.kubernetes.client.openapi.models.V1Node;
+import io.kubernetes.client.openapi.models.V1NodeAddress;
+import io.kubernetes.client.openapi.models.V1NodeStatus;
+import io.kubernetes.client.openapi.models.V1ObjectMeta;
+import io.kubernetes.client.openapi.models.V1Pod;
+import io.kubernetes.client.openapi.models.V1Service;
+import lombok.extern.slf4j.Slf4j;
@Slf4j
-public class K8SServiceRegistry {
+public class K8SServiceRegistry
+ implements KubernetesServiceListener, KubernetesPodListener,
+ KubernetesEndpointsListener, KubernetesNodeListener {
protected final Map<String/* ip */, ServiceMetaInfo> ipServiceMetaInfoMap;
protected final Map<String/* namespace:serviceName */, V1Service> idServiceMap;
@@ -60,14 +64,12 @@ public class K8SServiceRegistry {
protected final Map<String/* ip */, String/* namespace:serviceName */> ipServiceMap;
- protected final ExecutorService executor;
-
protected final ServiceNameFormatter serviceNameFormatter;
- protected final KubernetesNodeRegistry nodeRegistry;
-
private final EnvoyMetricReceiverConfig config;
+ private final Set<String> nodeIPs = Collections.newSetFromMap(new ConcurrentHashMap<>());
+
public K8SServiceRegistry(final EnvoyMetricReceiverConfig config) {
this.config = config;
@@ -76,142 +78,17 @@ public class K8SServiceRegistry {
idServiceMap = new ConcurrentHashMap<>();
ipPodMap = new ConcurrentHashMap<>();
ipServiceMap = new ConcurrentHashMap<>();
- executor = Executors.newCachedThreadPool(
- new ThreadFactoryBuilder()
- .setNameFormat("K8SServiceRegistry-%d")
- .setDaemon(true)
- .build()
- );
- nodeRegistry = new KubernetesNodeRegistry();
}
public void start() throws IOException {
- final ApiClient apiClient = Config.defaultClient();
- apiClient.setHttpClient(apiClient.getHttpClient()
- .newBuilder()
- .readTimeout(0, TimeUnit.SECONDS)
- .build());
- Configuration.setDefaultApiClient(apiClient);
-
- final CoreV1Api coreV1Api = new CoreV1Api();
- final SharedInformerFactory factory = new SharedInformerFactory(executor);
-
- // TODO: also listen to the EndpointSlice event after the client supports us to do so
- listenServiceEvents(coreV1Api, factory);
- listenEndpointsEvents(coreV1Api, factory);
- listenPodEvents(coreV1Api, factory);
-
- factory.startAllRegisteredInformers();
-
- nodeRegistry.start();
- }
-
- private void listenServiceEvents(final CoreV1Api coreV1Api, final SharedInformerFactory factory) {
- factory.sharedIndexInformerFor(
- params -> coreV1Api.listServiceForAllNamespacesCall(
- null,
- null,
- null,
- null,
- null,
- null,
- params.resourceVersion,
- null,
- params.timeoutSeconds,
- params.watch,
- null
- ),
- V1Service.class,
- V1ServiceList.class
- ).addEventHandler(new ResourceEventHandler<V1Service>() {
- @Override
- public void onAdd(final V1Service service) {
- addService(service);
- }
-
- @Override
- public void onUpdate(final V1Service oldService, final V1Service newService) {
- addService(newService);
- }
-
- @Override
- public void onDelete(final V1Service service, final boolean deletedFinalStateUnknown) {
- removeService(service);
- }
- });
- }
-
- private void listenEndpointsEvents(final CoreV1Api coreV1Api, final SharedInformerFactory factory) {
- factory.sharedIndexInformerFor(
- params -> coreV1Api.listEndpointsForAllNamespacesCall(
- null,
- null,
- null,
- null,
- null,
- null,
- params.resourceVersion,
- null,
- params.timeoutSeconds,
- params.watch,
- null
- ),
- V1Endpoints.class,
- V1EndpointsList.class
- ).addEventHandler(new ResourceEventHandler<V1Endpoints>() {
- @Override
- public void onAdd(final V1Endpoints endpoints) {
- addEndpoints(endpoints);
- }
-
- @Override
- public void onUpdate(final V1Endpoints oldEndpoints, final V1Endpoints newEndpoints) {
- addEndpoints(newEndpoints);
- }
-
- @Override
- public void onDelete(final V1Endpoints endpoints, final boolean deletedFinalStateUnknown) {
- removeEndpoints(endpoints);
- }
- });
- }
-
- private void listenPodEvents(final CoreV1Api coreV1Api, final SharedInformerFactory factory) {
- factory.sharedIndexInformerFor(
- params -> coreV1Api.listPodForAllNamespacesCall(
- null,
- null,
- null,
- null,
- null,
- null,
- params.resourceVersion,
- null,
- params.timeoutSeconds,
- params.watch,
- null
- ),
- V1Pod.class,
- V1PodList.class
- ).addEventHandler(new ResourceEventHandler<V1Pod>() {
- @Override
- public void onAdd(final V1Pod pod) {
- addPod(pod);
- }
-
- @Override
- public void onUpdate(final V1Pod oldPod, final V1Pod newPod) {
- addPod(newPod);
- }
-
- @Override
- public void onDelete(final V1Pod pod, final boolean deletedFinalStateUnknown) {
- removePod(pod);
- }
- });
+ KubernetesPodWatcher.INSTANCE.addListener(this).start();
+ KubernetesServiceWatcher.INSTANCE.addListener(this).start();
+ KubernetesEndpointWatcher.INSTANCE.addListener(this).start();
+ KubernetesNodeWatcher.INSTANCE.addListener(this).start();
}
- protected void addService(final V1Service service) {
+ @Override
+ public void onServiceAdded(final V1Service service) {
ofNullable(service.getMetadata()).ifPresent(
metadata -> idServiceMap.put(metadata.getNamespace() + ":" + metadata.getName(), service)
);
@@ -219,23 +96,37 @@ public class K8SServiceRegistry {
recompose();
}
- protected void removeService(final V1Service service) {
+ @Override
+ public void onServiceDeleted(final V1Service service) {
ofNullable(service.getMetadata()).ifPresent(
metadata -> idServiceMap.remove(metadata.getNamespace() + ":" + metadata.getName())
);
}
- protected void addPod(final V1Pod pod) {
+ @Override
+ public void onServiceUpdated(V1Service oldService, V1Service newService) {
+ onServiceAdded(newService);
+ }
+
+ @Override
+ public void onPodAdded(final V1Pod pod) {
ofNullable(pod.getStatus()).flatMap(status -> ofNullable(status.getPodIP())).ifPresent(podIP -> ipPodMap.put(podIP, pod));
recompose();
}
- protected void removePod(final V1Pod pod) {
+ @Override
+ public void onPodDeleted(final V1Pod pod) {
ofNullable(pod.getStatus()).flatMap(status -> ofNullable(status.getPodIP())).ifPresent(ipPodMap::remove);
}
- protected void addEndpoints(final V1Endpoints endpoints) {
+ @Override
+ public void onPodUpdated(V1Pod oldPod, V1Pod newPod) {
+ onPodAdded(newPod);
+ }
+
+ @Override
+ public void onEndpointsAdded(final V1Endpoints endpoints) {
V1ObjectMeta endpointsMetadata = endpoints.getMetadata();
if (isNull(endpointsMetadata)) {
log.error("Endpoints metadata is null: {}", endpoints);
@@ -254,7 +145,8 @@ public class K8SServiceRegistry {
recompose();
}
- protected void removeEndpoints(final V1Endpoints endpoints) {
+ @Override
+ public void onEndpointsDeleted(final V1Endpoints endpoints) {
ofNullable(endpoints.getSubsets()).ifPresent(subsets -> subsets.forEach(
subset -> ofNullable(subset.getAddresses()).ifPresent(addresses -> addresses.forEach(
address -> ofNullable(address.getIp()).ifPresent(ipServiceMap::remove)
@@ -262,6 +154,39 @@ public class K8SServiceRegistry {
));
}
+ @Override
+ public void onEndpointsUpdated(V1Endpoints oldEndpoints, V1Endpoints newEndpoints) {
+ onEndpointsAdded(newEndpoints);
+ }
+
+ @Override
+ public void onNodeAdded(final V1Node node) {
+ forEachAddress(node, nodeIPs::add);
+ }
+
+ @Override
+ public void onNodeUpdated(final V1Node oldNode, final V1Node newNode) {
+ onNodeAdded(newNode);
+ }
+
+ @Override
+ public void onNodeDeleted(final V1Node node) {
+ forEachAddress(node, nodeIPs::remove);
+ }
+
+ protected void forEachAddress(final V1Node node,
+ final Consumer<String> consume) {
+ Optional.ofNullable(node)
+ .map(V1Node::getStatus)
+ .map(V1NodeStatus::getAddresses)
+ .ifPresent(addresses ->
+ addresses.stream()
+ .map(V1NodeAddress::getAddress)
+ .filter(StringUtil::isNotBlank)
+ .forEach(consume)
+ );
+ }
+
protected List<ServiceMetaInfo.KeyValue> transformLabelsToTags(final V1ObjectMeta podMeta) {
final Map<String, String> labels = podMeta.getLabels();
final List<ServiceMetaInfo.KeyValue> tags = new ArrayList<>();
@@ -277,7 +202,7 @@ public class K8SServiceRegistry {
}
public ServiceMetaInfo findService(final String ip) {
- if (nodeRegistry.isNode(ip)) {
+ if (isNode(ip)) {
return config.serviceMetaInfoFactory().unknown();
}
final ServiceMetaInfo service = ipServiceMetaInfoMap.get(ip);
@@ -296,36 +221,42 @@ public class K8SServiceRegistry {
return;
}
- final Map<String, Object> context = ImmutableMap.of("service", service, "pod", pod);
- final V1ObjectMeta podMetadata = pod.getMetadata();
- if (isNull(podMetadata)) {
+ if (isNull(pod.getMetadata())) {
log.warn("Pod metadata is null, {}", pod);
return;
}
- ipServiceMetaInfoMap.computeIfAbsent(ip, unused -> {
- final ServiceMetaInfo serviceMetaInfo = new ServiceMetaInfo();
-
- try {
- serviceMetaInfo.setServiceName(serviceNameFormatter.format(context));
- } catch (Exception e) {
- log.error("Failed to evaluate service name.", e);
- final V1ObjectMeta serviceMetadata = service.getMetadata();
- if (isNull(serviceMetadata)) {
- log.warn("Service metadata is null, {}", service);
- return config.serviceMetaInfoFactory().unknown();
- }
- serviceMetaInfo.setServiceName(serviceMetadata.getName());
- }
- serviceMetaInfo.setServiceInstanceName(
- String.format("%s.%s", podMetadata.getName(), podMetadata.getNamespace()));
- serviceMetaInfo.setTags(transformLabelsToTags(podMetadata));
-
- return serviceMetaInfo;
- });
+ ipServiceMetaInfoMap.computeIfAbsent(ip, unused -> composeServiceMetaInfo(service, pod));
});
}
+ protected ServiceMetaInfo composeServiceMetaInfo(final V1Service service, final V1Pod pod) {
+ final Map<String, Object> context = ImmutableMap.of("service", service, "pod", pod);
+ final ServiceMetaInfo serviceMetaInfo = new ServiceMetaInfo();
+ final V1ObjectMeta podMetadata = pod.getMetadata();
+
+ try {
+ serviceMetaInfo.setServiceName(serviceNameFormatter.format(context));
+ } catch (Exception e) {
+ log.error("Failed to evaluate service name.", e);
+ final V1ObjectMeta serviceMetadata = service.getMetadata();
+ if (isNull(serviceMetadata)) {
+ log.warn("Service metadata is null, {}", service);
+ return config.serviceMetaInfoFactory().unknown();
+ }
+ serviceMetaInfo.setServiceName(serviceMetadata.getName());
+ }
+ serviceMetaInfo.setServiceInstanceName(
+ String.format("%s.%s", podMetadata.getName(), podMetadata.getNamespace()));
+ serviceMetaInfo.setTags(transformLabelsToTags(podMetadata));
+
+ return serviceMetaInfo;
+ }
+
+ public boolean isNode(final String ip) {
+ return nodeIPs.contains(ip);
+ }
+
public boolean isEmpty() {
return ipServiceMetaInfoMap.isEmpty();
}
diff --git a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/k8s/KubernetesNodeRegistry.java b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/k8s/KubernetesNodeRegistry.java
deleted file mode 100644
index 6a9983b6b0..0000000000
--- a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/k8s/KubernetesNodeRegistry.java
+++ /dev/null
@@ -1,129 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- *
- */
-
-package org.apache.skywalking.oap.server.receiver.envoy.als.k8s;
-
-import com.google.common.util.concurrent.ThreadFactoryBuilder;
-import io.kubernetes.client.informer.ResourceEventHandler;
-import io.kubernetes.client.informer.SharedInformerFactory;
-import io.kubernetes.client.openapi.ApiClient;
-import io.kubernetes.client.openapi.Configuration;
-import io.kubernetes.client.openapi.apis.CoreV1Api;
-import io.kubernetes.client.openapi.models.V1Node;
-import io.kubernetes.client.openapi.models.V1NodeAddress;
-import io.kubernetes.client.openapi.models.V1NodeList;
-import io.kubernetes.client.openapi.models.V1NodeStatus;
-import io.kubernetes.client.util.Config;
-import java.io.IOException;
-import java.util.Collections;
-import java.util.Optional;
-import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.TimeUnit;
-import java.util.function.Consumer;
-import lombok.extern.slf4j.Slf4j;
-import org.apache.skywalking.oap.server.library.util.StringUtil;
-
-@Slf4j
-public final class KubernetesNodeRegistry implements ResourceEventHandler<V1Node> {
- private final Set<String> nodeIPs;
-
- private final ExecutorService executor;
-
- public KubernetesNodeRegistry() {
- nodeIPs = Collections.newSetFromMap(new ConcurrentHashMap<>());
- executor = Executors.newCachedThreadPool(
- new ThreadFactoryBuilder()
- .setNameFormat("KubernetesNodeRegistry-%d")
- .setDaemon(true)
- .build()
- );
- }
-
- public void start() throws IOException {
- final ApiClient apiClient = Config.defaultClient();
- apiClient.setHttpClient(apiClient.getHttpClient()
- .newBuilder()
- .readTimeout(0, TimeUnit.SECONDS)
- .build());
- Configuration.setDefaultApiClient(apiClient);
-
- final CoreV1Api coreV1Api = new CoreV1Api();
- final SharedInformerFactory factory = new SharedInformerFactory(executor);
-
- listenNodeEvents(coreV1Api, factory);
-
- factory.startAllRegisteredInformers();
- }
-
- private void listenNodeEvents(final CoreV1Api coreV1Api,
- final SharedInformerFactory factory) {
- factory.sharedIndexInformerFor(
- params -> coreV1Api.listNodeCall(
- null,
- null,
- null,
- null,
- null,
- null,
- params.resourceVersion,
- null,
- params.timeoutSeconds,
- params.watch,
- null
- ),
- V1Node.class,
- V1NodeList.class
- ).addEventHandler(this);
- }
-
- @Override
- public void onAdd(final V1Node node) {
- forEachAddress(node, nodeIPs::add);
- }
-
- @Override
- public void onUpdate(final V1Node oldNode, final V1Node newNode) {
- onAdd(newNode);
- }
-
- @Override
- public void onDelete(final V1Node node,
- final boolean deletedFinalStateUnknown) {
- forEachAddress(node, nodeIPs::remove);
- }
-
- public void forEachAddress(final V1Node node,
- final Consumer<String> consume) {
- Optional.ofNullable(node)
- .map(V1Node::getStatus)
- .map(V1NodeStatus::getAddresses)
- .ifPresent(addresses ->
- addresses.stream()
- .map(V1NodeAddress::getAddress)
- .filter(StringUtil::isNotBlank)
- .forEach(consume)
- );
- }
-
- public boolean isNode(final String ip) {
- return nodeIPs.contains(ip);
- }
-}