You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by "wu-sheng (via GitHub)" <gi...@apache.org> on 2026/10/14 02:51:41 UTC
[PR] Serve Kubernetes metadata from watch-maintained local stores instead of API calls on gRPC threads [skywalking]
wu-sheng opened a new pull request, #14135:
URL: https://github.com/apache/skywalking/pull/14135
### Fix Kubernetes metadata lookups calling the API server on gRPC handler threads
- [x] Add a unit test to verify that the fix works.
- [x] Explain briefly why the bug exists and how to fix it.
**Why.** Every IP → pod / service / Istio ServiceEntry lookup went through Guava `LoadingCache`s in `library-kubernetes-support` whose loaders made synchronous Kubernetes REST calls on the caller's thread: a cluster-wide `pods?fieldSelector=status.podIP` list per IP, service GETs, and whole-cluster endpoints / services / nodes / ServiceEntry lists under single cache keys. The callers are the gRPC handler threads of Envoy ALS (`k8s-mesh`) and of the eBPF access log, and MAL `retagByK8sMeta` / LAL and MAL process naming through `K8sInfoRegistry`. Each cache miss, each `expireAfterWrite` expiry, and each failed call (failures were not cached) repeated the call on that thread. When the API server is slow or unreachable, every access log stream touching an uncached IP holds a handler thread, with the batch it was analyzing, for up to the client timeout; gRPC's default executor is unbounded before JDK 25, so threads and retained batches pile up until OAP runs out of memory.
The Istio ServiceEntry path added its own: hostname resolutions were created from the lookup, a DNS group per host was kept forever, `withMaxAttempts(3)` made Armeria retry with a `-1` delay (an immediate retry, forever) once exhausted, and nested `parallelStream`s ran on the common pool.
**Fix.**
- `library-kubernetes-support` now keeps one cluster-wide, watch-maintained store per resource kind (pods, services, endpoints, nodes, Istio ServiceEntries), built on fabric8 informers (`ResourceInformer`). Lookups are pure local reads: what is not in the store yet is unknown, never waited for. A read that finds the watch not started hands the start to a background thread; the Envoy ALS registries start their watches at module boot. Listing and watching retry forever (fabric8's default handler stops an informer for good on a failed first list or a non-410 watch error). Initial lists are paged.
- Stores keep only the fields the lookups read (name, namespace, labels, IPs, selectors, the matched ServiceEntry fields). The fields a `k8sServiceNameRule` / `istioServiceNameRule` reads are parsed from the rule at boot and kept too, mapped from bean properties onto the serialized fields holding them, so custom rules render as before.
- Matches are ordered by namespace and name, so every OAP node picks the same object; pod-by-IP prefers a pod that has not terminated (a completed pod keeps its IP after it is reused).
- `K8SServiceRegistry`, `IstioServiceEntryRegistry` and `K8sInfoRegistry` keep their per-IP caches but write them only after the stores have completed their first list, so a startup miss is not cached for three minutes.
- ServiceEntry hostname resolutions start from ServiceEntry events, in the background, with Armeria's default (unlimited, growing) backoff. They are kept until OAP stops: a closed `DnsEndpointGroup` stays registered in Armeria's shared DNS cache, so closing and recreating them would leak.
- Docs: `als_setting.md` and `mal.md` state the permissions (`get`, `list`, `watch` on pods, services, endpoints, nodes and `serviceentries.networking.istio.io`; the Helm chart already grants them) and the startup behaviour.
**Tests.**
- `ResourceInformerTest` drives a real fabric8 informer through a fake API server: reads never wait for a first list that never completes, the start runs on a background thread, 403s are retried until the list succeeds, watch events move the index, a 410 re-list serves the old state until replaced, a broken watch keeps retrying, stored copies are trimmed, results are ordered, kept fields are fixed once started, and a client that cannot be built is not retried.
- `ResourceTrimmerTest` checks, with the production trimmer and `PropertyUtils`, that the properties naming rules read have the same value on the stored copy as on the full object, including derived properties and `additionalProperties`. `LookupFieldsTest` covers the index values and trimmed copies of each store.
- `K8SServiceRegistryTest`, `IstioServiceEntryRegistryTest` and `ServiceNameFormatterFieldPathsTest` cover the registries and rule parsing. Existing tests (`K8sTagTest`, `MALExpressionExecutionTest`, `K8SALSServiceMeshHTTPAnalysisTest`, `FieldFormatterTest`) pass unchanged.
- [ ] If this pull request closes/resolves/fixes an existing issue, replace the issue number. Closes #<issue number>.
- [x] Update the [`CHANGES` log](https://github.com/apache/skywalking/blob/master/docs/en/changes/changes.md).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Read Kubernetes metadata in the background, off the threads analyzing ALS, eBPF access logs and MAL [skywalking]
Posted by "wu-sheng (via GitHub)" <gi...@apache.org>.
wu-sheng commented on PR #14135:
URL: https://github.com/apache/skywalking/pull/14135#issuecomment-6184502575
## Updated lookup rules
This revision changes how the Kubernetes metadata lookups retry, cache and wait. The PR description is updated to match.
**What changed**
- **A failed read is no longer cached like "nothing found".** A failure (403, 5xx, the API server not answering or refusing the connection) is retried on the first use 3 minutes later. A read that succeeds and matches nothing is read again on the first use 1 minute later, as the key may be new to the API server or to the lists OAP matched it against.
- **New pods are found within about a minute.** A pod found by IP but not in the endpoints read (Envoy ALS), or selected by no service in the services read (MAL), makes the reader read that list again once it is older than 1 minute. Before, the 3-minute lists could hide a new pod for 3–6 minutes.
- **Cached data lives longer.** It is still read again on its first use 3 minutes after it was written, but is now dropped 21 minutes after its last write (7 refresh periods), per cache: `als` for Envoy ALS, `mal` for MAL and the eBPF access log. A read again that finds nothing or fails keeps the data and is retried a refresh period later.
- **MAL waits at most 2 seconds per `retagByK8sMeta`, not per pod.** The step queues the reads of all its pods first, then waits for them within one `malMetadataReadTimeout`, now 2000 ms by default. A pod's wait counts from when a step first waited for it, so the several `retagByK8sMeta` of one Export over the same pods wait 2 s at most in all, whatever happens to their reads meanwhile (a full lane admitting them later, an entry expiring). A cached pod is not waited for, even while it is read again. Its lookups after that never wait. That keeps an Export under the OpenTelemetry Collector's 5-second export timeout, after which it sends the data again. Pods not read by then get an empty service for this Export, and are read in the background for the next one.
- **Failed lists are not read once per IP.** A list the API server does not serve (404, e.g. no Istio) is an empty list. A failed list read keeps the list read before; with none, the failure is kept. After either, the list is not read again for 3 minutes, even for a pod not in it. So a 403 on ServiceEntries is read once, not once per new IP.
- **An eBPF connection resolves its unknown addresses again.** The later messages of a connection do not carry it again; a connection whose addresses were unknown now resolves them again on those messages once their metadata is read, instead of staying unknown (or dropped as same-service traffic) for the rest of the stream. A resolved address is kept.
- **Every setting has a code default and is written out in `application.yml` with the same value.** An invalid value, including a bare number for an `als=…,mal=…` key, stops OAP at boot with a message showing the form.
**Lookup rules** (defaults; keys in `core.default`)
| | Envoy ALS `k8s-mesh` (and Istio ServiceEntries), eBPF access log | MAL `retagByK8sMeta`, process naming |
|---|---|---|
| Key not cached | unknown now; the read is queued (`K8sLookup` lane); an eBPF connection resolves it again on its later messages | all the step's reads queued first (`K8sLookup-waiting` lane), then waited for up to `malMetadataReadTimeout` (2000 ms) in all, the steps of one Export together |
| Data cached | the data; read again in the background on the first use `k8sMetadataRefreshPeriod` (3 min) after it was written | the same |
| Nothing found | unknown; read again on the first use `k8sMetadataNotFoundRetryPeriod` (1 min) after it was written; endpoints read again for a pod not in them once that old | empty service; the same, with the services list |
| Read failed (403, 5xx, not answering, refused) | unknown; retried on the first use `k8sMetadataRefreshPeriod` (3 min) after it failed | empty service; the same |
| Read again of data finds nothing or fails | data kept; retried on a use a refresh period later | the same |
| Dropped | `k8sMetadataCacheTime` `als` (21 min) after the last write | `k8sMetadataCacheTime` `mal` (21 min) after the last write |
| Lane full (`k8sMetadataReadQueueSize`) | unknown, not cached; queued again by the next lookup | empty service at once; the same |
**Settings**
| Key | Env | Default |
|---|---|---|
| `malMetadataReadTimeout` | `SW_CORE_MAL_METADATA_READ_TIMEOUT` | `2000` ms |
| `k8sMetadataCacheTime` | `SW_CORE_K8S_METADATA_CACHE_TIME` | `als=21,mal=21` minutes |
| `k8sMetadataRefreshPeriod` | `SW_CORE_K8S_METADATA_REFRESH_PERIOD` | `3` minutes |
| `k8sMetadataNotFoundRetryPeriod` | `SW_CORE_K8S_METADATA_NOT_FOUND_RETRY_PERIOD` | `1` minute |
| `k8sMetadataReadQueueSize` | `SW_CORE_K8S_METADATA_READ_QUEUE_SIZE` | empty (no limit) |
Two reader threads serve all of it, one per lane. Each runs one read at a time, is created on demand and ends after 60 s idle (a virtual thread on JDK 25+). Refreshes are triggered by use, not by a timer.
**Verified against a real API server** (kind, with the API server's audit log as the record of each request OAP sent):
- **New pod:** a new pod's IP was first looked up 1 s after the pod started, and found nothing, as the endpoints read 24 s earlier lacked it. The use 66 s later read the endpoints again and resolved it.
- **403:** nothing was read again at +5 s or +75 s; everything was read again at +186 s. ServiceEntries were listed once, not once per IP.
- **MAL with the API server unresponsive:** one Export of 50 unread pods, through four `retagByK8sMeta`, took 2.2 s; the same pods again took 0.2 s; 41 cached pods took 0.2 s. With `mal=1`, 6 unread pods took 2.3 s.
- **Short cache (1-minute refresh, 2-minute cache time):** pods in use were read again at +66 s and +129 s and stayed cached. Pods unused for 2.5 minutes had expired.
- **Envoy ALS never waited:** 0.2–0.6 s per batch, healthy, paused or 403, on JDK 11 and 25.
- **Load with the API server unresponsive:** 2000 unread IPs over 50 streams, JDK 11. No thread grew; only the two readers were in Kubernetes code.
- **Boot:** every invalid setting stopped OAP with a message showing the expected form.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Read Kubernetes metadata in the background, off the threads analyzing ALS, eBPF access logs and MAL [skywalking]
Posted by "wu-sheng (via GitHub)" <gi...@apache.org>.
wu-sheng merged PR #14135:
URL: https://github.com/apache/skywalking/pull/14135
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Serve Kubernetes metadata from watch-maintained local stores instead of API calls on gRPC threads [skywalking]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #14135:
URL: https://github.com/apache/skywalking/pull/14135#discussion_r4266361317
##########
oap-server/analyzer/meter-analyzer/src/main/java/org/apache/skywalking/oap/meter/analyzer/v2/k8s/K8sInfoRegistry.java:
##########
@@ -130,24 +72,102 @@ public String findServiceName(String namespace, String podName) {
@SneakyThrows
public ObjectID findService(String namespace, String podName) {
- return this.podServiceMap.get(
- ObjectID
- .builder()
- .name(podName)
- .namespace(namespace)
- .build());
+ final ObjectID pod = ObjectID.builder().name(podName).namespace(namespace).build();
+ if (!KubernetesPods.INSTANCE.hasSynced() || !KubernetesServices.INSTANCE.hasSynced()) {
+ return resolveServiceOfPod(pod);
+ }
+ return this.podServiceMap.get(pod);
}
@SneakyThrows
public ObjectID findPodByIP(String ip) {
+ if (!KubernetesPods.INSTANCE.hasSynced()) {
+ return resolvePodByIP(ip);
+ }
return this.ipPodMap.get(ip);
}
@SneakyThrows
public ObjectID findServiceByIP(String ip) {
+ if (!KubernetesServices.INSTANCE.hasSynced()) {
+ return resolveServiceByIP(ip);
+ }
return this.ipServiceMap.get(ip);
}
+ private static ObjectID resolvePodByIP(final String ip) {
+ return KubernetesPods.INSTANCE
+ .findByIP(ip)
+ .map(it -> ObjectID
+ .builder()
+ .name(it.getMetadata().getName())
+ .namespace(it.getMetadata().getNamespace())
+ .build())
+ .orElse(ObjectID.EMPTY);
+ }
+
+ private static ObjectID resolveServiceByIP(final String ip) {
+ return KubernetesServices.INSTANCE
+ .list()
+ .stream()
+ .filter(it -> it.getSpec() != null)
+ .filter(it -> it.getStatus() != null)
+ .filter(it -> it.getMetadata() != null)
+ .filter(it -> (it.getSpec().getClusterIPs() != null &&
+ it.getSpec().getClusterIPs().stream()
+ .anyMatch(clusterIP -> Objects.equals(clusterIP, ip)))
+ || (it.getStatus().getLoadBalancer() != null &&
+ it.getStatus().getLoadBalancer().getIngress() != null &&
+ it.getStatus().getLoadBalancer().getIngress().stream()
+ .anyMatch(ingress -> Objects.equals(ingress.getIp(), ip))))
+ .map(it -> ObjectID
+ .builder()
+ .name(it.getMetadata().getName())
+ .namespace(it.getMetadata().getNamespace())
+ .build())
+ .findFirst()
+ .orElse(ObjectID.EMPTY);
+ }
Review Comment:
This is still an O(#services) scan for each resolve (including during startup when `hasSynced()` is false and misses aren’t cached). Now that `KubernetesServices` is informer-backed, it’d be more efficient to index services by “address IP” (clusterIPs + loadBalancer ingress IPs) inside `KubernetesServices` and expose a `findByAddressIP(ip)` method, so this lookup becomes an indexed read instead of a full scan.
##########
oap-server/server-library/library-kubernetes-support/src/main/java/org/apache/skywalking/library/kubernetes/KubernetesPods.java:
##########
@@ -19,61 +19,119 @@
package org.apache.skywalking.library.kubernetes;
-import com.google.common.cache.CacheBuilder;
-import com.google.common.cache.CacheLoader;
-import com.google.common.cache.LoadingCache;
import io.fabric8.kubernetes.api.model.Pod;
-import lombok.SneakyThrows;
+import io.fabric8.kubernetes.api.model.PodBuilder;
+import io.fabric8.kubernetes.api.model.PodIP;
+import io.fabric8.kubernetes.api.model.PodStatus;
+import io.fabric8.kubernetes.api.model.PodStatusBuilder;
+import io.fabric8.kubernetes.client.informers.SharedIndexInformer;
-import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
import java.util.Optional;
+/**
+ * Pods of the whole cluster, served from a watch-maintained local store. Lookups never call the API server; until
+ * {@link #hasSynced()}, they see only the pods listed so far.
+ */
public enum KubernetesPods {
INSTANCE;
- private final LoadingCache<String, Optional<Pod>> podByIP;
- private final LoadingCache<ObjectID, Optional<Pod>> podByObjectID;
+ private static final String IP_INDEX = "podIP";
- @SneakyThrows
- KubernetesPods() {
- final CacheBuilder<Object, Object> cacheBuilder =
- CacheBuilder.newBuilder()
- .expireAfterWrite(Duration.ofMinutes(5));
+ private final ResourceInformer<Pod> informer = new ResourceInformer<>(
+ "pods",
+ KubernetesPods::createInformer,
+ new ResourceTrimmer<>(
+ Pod.class, KubernetesPods::keepLookupFields,
+ "metadata.name", "metadata.namespace", "metadata.labels",
+ "status.phase", "status.podIP", "status.podIPs"
+ ),
+ Map.of(IP_INDEX, KubernetesPods::podIPs)
+ );
- podByIP = cacheBuilder.build(new CacheLoader<>() {
- @Override
- public Optional<Pod> load(String ip) {
- return SharedKubernetesClient.INSTANCE.get()
- .pods()
- .inAnyNamespace()
- .withField("status.podIP", ip)
- .list()
- .getItems()
- .stream()
- .findFirst();
- }
- });
+ public Optional<Pod> findByIP(final String ip) {
+ return choosePod(informer.listByIndex(IP_INDEX, ip));
+ }
+
+ public Optional<Pod> findByObjectID(final ObjectID id) {
+ return informer.findByKey(id.namespace(), id.name());
+ }
+
+ /**
+ * @return true once the pods of the whole cluster have been listed. Before that, lookups may miss pods, and
+ * callers should not cache their misses.
+ */
+ public boolean hasSynced() {
+ return informer.hasSynced();
+ }
- podByObjectID = cacheBuilder.build(new CacheLoader<>() {
- @Override
- public Optional<Pod> load(ObjectID objectID) {
- return Optional.ofNullable(
- SharedKubernetesClient.INSTANCE.get()
- .pods()
- .inNamespace(objectID.namespace())
- .withName(objectID.name())
- .get());
+ /**
+ * Starts watching pods now instead of on the first lookup.
+ *
+ * @param extraFieldPaths JSON paths, such as {@code spec.serviceAccountName}, to keep on top of the name,
+ * namespace, labels, phase and IPs. Register them before the first lookup.
+ */
+ public void startWatching(final Collection<String> extraFieldPaths) {
+ informer.startWatching(extraFieldPaths);
Review Comment:
The API/docs here say callers should pass “JSON paths”, but the trimming logic (`ResourceTrimmer`) is explicitly based on *bean property paths* and then maps them to serialized field paths. This mismatch can cause callers to pass the wrong “path vocabulary” for properties whose bean name differs from the serialized field name or for derived properties. Recommend aligning the public docs and parameter naming across stores/`ResourceInformer` to “bean property paths (as emitted by the rule parser)”, or otherwise explicitly document/accept both forms.
##########
oap-server/server-library/library-kubernetes-support/src/main/java/org/apache/skywalking/library/kubernetes/ResourceInformer.java:
##########
@@ -0,0 +1,234 @@
+/*
+ * 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.fabric8.kubernetes.api.model.HasMetadata;
+import io.fabric8.kubernetes.client.informers.ResourceEventHandler;
+import io.fabric8.kubernetes.client.informers.SharedIndexInformer;
+import io.fabric8.kubernetes.client.informers.cache.Cache;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.skywalking.oap.server.library.util.VirtualThreads;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Function;
+import java.util.function.Supplier;
+
+/**
+ * One cluster-wide, watch-maintained local store of a Kubernetes resource kind.
+ *
+ * <p>Reads are pure local lookups and never wait: they return what the store holds, which fills page by page during
+ * the first list and is then kept up to date by the watch in the background. A read that finds the watch not started
+ * hands the start to a background thread and returns at once; {@link #startWatching(Collection)} starts it on the
+ * caller's thread, for module boot. Listing and watching are retried forever after failures.
+ */
+@Slf4j
+final class ResourceInformer<T extends HasMetadata> {
+ /**
+ * Page size of the initial list and of re-lists, so a large cluster is never buffered as one response.
+ */
+ static final long LIST_PAGE_SIZE = 500L;
+
+ /**
+ * Results are sorted so that every OAP replica picks the same object when several match. The informer's index
+ * returns keys in hash order, while the API server lists objects ordered by namespace and name.
+ */
+ static final Comparator<HasMetadata> BY_NAMESPACE_AND_NAME =
+ Comparator.comparing((HasMetadata it) -> nullToEmpty(it.getMetadata().getNamespace()))
+ .thenComparing(it -> nullToEmpty(it.getMetadata().getName()));
+
+ private static final long CREATE_RETRY_INTERVAL_MILLIS = TimeUnit.MINUTES.toMillis(1);
+
+ /**
+ * Starts the watches that reads asked for, so a request thread never builds the client or an informer itself.
+ * It keeps no thread while idle.
+ */
+ private static final ExecutorService BACKGROUND_STARTER = VirtualThreads.createExecutor(
+ "K8sInformer-start",
+ () -> new ThreadPoolExecutor(0, 1, 60, TimeUnit.SECONDS, new LinkedBlockingQueue<>(), r -> {
+ final Thread thread = new Thread(r, "K8sInformer-start");
+ thread.setDaemon(true);
+ return thread;
+ })
+ );
+
+ private final String resourceName;
+ private final Supplier<SharedIndexInformer<T>> informerFactory;
+ private final ResourceTrimmer<T> trimmer;
+ private final Map<String, Function<T, List<String>>> indexers;
+ private final List<ResourceEventHandler<? super T>> eventHandlers = new ArrayList<>();
+ private final Set<String> extraFieldPaths = new HashSet<>();
+ private final AtomicBoolean backgroundStartPending = new AtomicBoolean();
+
+ private volatile SharedIndexInformer<T> informer;
+ private volatile long nextCreateAttemptMillis;
+
+ ResourceInformer(final String resourceName,
+ final Supplier<SharedIndexInformer<T>> informerFactory,
+ final ResourceTrimmer<T> trimmer,
+ final Map<String, Function<T, List<String>>> indexers) {
+ this.resourceName = resourceName;
+ this.informerFactory = informerFactory;
+ this.trimmer = trimmer;
+ this.indexers = indexers;
+ }
+
+ /**
+ * Starts watching on the caller's thread if not started yet, keeping {@code fieldPaths} (JSON paths such as
+ * {@code spec.nodeName}) on top of the fields the lookups need. The kept fields are fixed once the watch starts,
+ * so register them at module boot, before any lookup.
+ */
+ synchronized void startWatching(final Collection<String> fieldPaths) {
+ if (informer == null) {
+ extraFieldPaths.addAll(fieldPaths);
+ } else {
+ final Set<String> kept = trimmer.uncoveredPaths(extraFieldPaths);
+ final Set<String> missing = trimmer.uncoveredPaths(fieldPaths);
+ missing.removeIf(path -> kept.stream().anyMatch(
+ keptPath -> keptPath.isEmpty() || path.equals(keptPath) || path.startsWith(keptPath + ".")));
+ if (!missing.isEmpty()) {
+ log.warn("The {} store started before the fields {} were registered, so they are not kept; a naming"
+ + " rule reading them renders '-' for them.", resourceName, missing);
+ }
+ }
+ startWatching();
+ }
+
+ /**
+ * Starts watching on the caller's thread if not started yet. A failed start is retried at most once per
+ * {@link #CREATE_RETRY_INTERVAL_MILLIS}.
+ */
+ synchronized void startWatching() {
+ final long now = System.currentTimeMillis();
+ if (informer != null || now < nextCreateAttemptMillis) {
+ return;
+ }
+ nextCreateAttemptMillis = now + CREATE_RETRY_INTERVAL_MILLIS;
+ try {
+ final SharedIndexInformer<T> created = informerFactory.get();
+ created.itemStore(new TrimmingItemStore<>(trimmer.keeping(extraFieldPaths)));
+ created.addIndexers(indexers);
+ // fabric8's default handler stops the informer for good on a failed first list, and on a watch error
+ // other than HTTP 410, freezing the store. fabric8 logs every failure and retries with backoff.
+ created.exceptionHandler((isStarted, t) -> true);
+ eventHandlers.forEach(created::addEventHandler);
+ // Published before starting: event handlers of the first list may read the store through it.
+ informer = created;
+ // Not joined: the first list completes in the background, reads see what has been stored so far.
+ created.start();
+ } catch (final RuntimeException e) {
+ informer = null;
+ log.warn("Failed to start watching {}, retrying in {} seconds; lookups find nothing until then.",
+ resourceName, TimeUnit.MILLISECONDS.toSeconds(CREATE_RETRY_INTERVAL_MILLIS), e);
+ } catch (final LinkageError e) {
+ // The shared client's initialization failed, e.g. on an invalid Kubernetes configuration; a class
+ // initialization is never retried by the JVM.
+ informer = null;
+ nextCreateAttemptMillis = Long.MAX_VALUE;
+ log.error("Failed to create the Kubernetes client, {} lookups find nothing until OAP restarts.",
+ resourceName, e);
+ }
+ }
+
+ /**
+ * @return true once the first full list has been stored. It stays true while a later re-list or a broken watch
+ * is retried, as the last stored state remains the best answer.
+ */
+ boolean hasSynced() {
+ final SharedIndexInformer<T> current = currentInformer();
+ return current != null && current.hasSynced();
+ }
+
+ List<T> listByIndex(final String indexName, final String indexValue) {
+ final SharedIndexInformer<T> current = currentInformer();
+ if (current == null || indexValue == null) {
+ return Collections.emptyList();
+ }
+ final List<T> found = new ArrayList<>(current.getIndexer().byIndex(indexName, indexValue));
+ found.sort(BY_NAMESPACE_AND_NAME);
+ return found;
+ }
+
+ Optional<T> findByKey(final String namespace, final String name) {
+ final SharedIndexInformer<T> current = currentInformer();
+ if (current == null || name == null) {
+ return Optional.empty();
+ }
+ return Optional.ofNullable(current.getIndexer().getByKey(Cache.namespaceKeyFunc(namespace, name)));
+ }
+
+ List<T> list() {
+ final SharedIndexInformer<T> current = currentInformer();
+ if (current == null) {
+ return Collections.emptyList();
+ }
+ final List<T> all = new ArrayList<>(current.getIndexer().list());
+ all.sort(BY_NAMESPACE_AND_NAME);
+ return all;
+ }
+
+ /**
+ * Registers a handler, called after the store has been updated, on the Kubernetes client's informer executor.
+ * A handler registered after the store started is first called on the registering thread for every stored
+ * object.
+ */
+ synchronized void addEventHandler(final ResourceEventHandler<? super T> handler) {
+ eventHandlers.add(handler);
+ if (informer != null) {
Review Comment:
The Javadoc states that if a handler is registered after the store started, it is “first called on the registering thread for every stored object”, but the implementation only calls `informer.addEventHandler(handler)` and does not replay existing items. Either update the documentation to reflect actual behavior, or implement the documented replay (e.g., iterate currently stored items and invoke `handler.onAdd` for each on the registering thread) so downstream code can safely rely on this contract.
##########
oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/istio/IstioServiceEntryRegistry.java:
##########
@@ -54,85 +61,142 @@ public class IstioServiceEntryRegistry {
protected final LoadingCache<String/* ip */, ServiceMetaInfo> ipServiceMetaInfoMap;
protected final Map<String, DnsAddressEndpointGroup> hostnameResolvers = new ConcurrentHashMap<>();
- @SneakyThrows
+ private final Set<String> ignoredNamespaces;
+
public IstioServiceEntryRegistry(final EnvoyMetricReceiverConfig config) {
this.config = config;
serviceNameFormatter = new ServiceNameFormatter(config.getIstioServiceNameRule());
+ ignoredNamespaces = config.getIstioServiceEntryIgnoredNamespaces();
- final var ignoredNamespaces = config.getIstioServiceEntryIgnoredNamespaces();
+ ipServiceMetaInfoMap = CacheBuilder.newBuilder()
+ .expireAfterWrite(Duration.ofMinutes(3))
+ .build(CacheLoader.from(this::resolveServiceMetaInfo));
- final var cacheBuilder = CacheBuilder.newBuilder().expireAfterWrite(Duration.ofMinutes(3));
+ // Hostname resolutions start in the background as ServiceEntries are stored, never on a lookup.
+ IstioServiceEntries.INSTANCE.addEventHandler(new ResourceEventHandler<ServiceEntry>() {
+ @Override
+ public void onAdd(final ServiceEntry serviceEntry) {
+ createHostnameResolvers(serviceEntry);
+ }
+
+ @Override
+ public void onUpdate(final ServiceEntry oldServiceEntry, final ServiceEntry newServiceEntry) {
+ createHostnameResolvers(newServiceEntry);
+ }
- ipServiceMetaInfoMap = cacheBuilder.build(new CacheLoader<>() {
@Override
- public ServiceMetaInfo load(String ip) {
- final var serviceEntry = IstioServiceEntries
- .INSTANCE
- .list()
- .parallelStream()
- .filter(se -> se.getMetadata() != null)
- .filter(se -> se.getSpec() != null)
- .filter(se -> !ignoredNamespaces.contains(se.getMetadata().getNamespace()))
- .filter(se -> {
- final var spec = se.getSpec();
- if (spec.getResolution() == null) {
- log.debug("Unsupported service entry resolution: {}", spec.getResolution());
- return false;
- }
- switch (spec.getResolution()) {
- case STATIC:
- return spec
- .getAddresses()
- .parallelStream()
- .anyMatch(address -> {
- if (address.contains("/")) { // CIDR
- final var subnet = new SubnetUtils(address);
- return subnet.getInfo().isInRange(ip);
- }
- return Objects.equals(ip, address);
- }) ||
- spec
- .getEndpoints()
- .parallelStream()
- .map(WorkloadEntrySpec::getAddress)
- .anyMatch(address -> Objects.equals(address, ip));
- case DNS:
- case DNS_ROUND_ROBIN:
- return spec
- .getHosts()
- .parallelStream()
- .map(host -> hostnameResolvers.computeIfAbsent(host, it ->
- DnsAddressEndpointGroup.builder(it)
- .backoff(Backoff.exponential(1000, 32000).withJitter(0.2).withMaxAttempts(3))
- .build()
- ))
- .anyMatch(dnsAddressEndpointGroup -> {
- if (dnsAddressEndpointGroup.whenReady().isDone()) {
- return dnsAddressEndpointGroup
- .endpoints()
- .parallelStream()
- .anyMatch(endpoint -> Objects.equals(endpoint.ipAddr(), ip));
- }
- return false;
- });
- default:
- log.debug("Unsupported service entry resolution: {}", spec.getResolution());
- return false;
- }
- })
- .findFirst();
-
- if (serviceEntry.isEmpty()) {
- log.debug("No corresponding service entry for IP: {}", ip);
- return config.serviceMetaInfoFactory().unknown();
- }
-
- log.debug(
- "Composing service meta info from service entry for IP: {}", ip);
- return composeServiceMetaInfo(serviceEntry.get(), ip);
+ public void onDelete(final ServiceEntry serviceEntry, final boolean deletedFinalStateUnknown) {
+ // Resolutions are kept, see createHostnameResolver.
}
});
+ // Starts the watch at boot, keeping every ServiceEntry field the naming rule reads.
+ IstioServiceEntries.INSTANCE.startWatching(serviceNameFormatter.fieldPathsOf("serviceEntry"));
+ }
+
+ /**
+ * Starts resolving the hostnames of {@code serviceEntry} that are not resolved yet, when it is resolved by DNS.
+ */
+ protected void createHostnameResolvers(final ServiceEntry serviceEntry) {
+ if (!isConsidered(serviceEntry)
+ || !isResolvedByDns(serviceEntry.getSpec())
+ || serviceEntry.getSpec().getHosts() == null) {
+ return;
+ }
+ serviceEntry.getSpec()
+ .getHosts()
+ .stream()
+ // A wildcard host never resolves.
+ .filter(host -> host != null && !host.isEmpty() && !host.contains("*"))
+ .forEach(host -> hostnameResolvers.computeIfAbsent(host, this::createHostnameResolver));
+ }
+
+ /**
+ * Creates the resolution of {@code host}, kept until OAP stops. Armeria's default backoff retries a failing
+ * resolution with growing delays; an attempt-limited backoff would instead retry without delay once exhausted.
+ * Closing a resolution does not release it: it stays registered in Armeria's shared DNS cache, so a host that
+ * comes back reuses its resolution instead of leaking a new one.
+ */
+ protected DnsAddressEndpointGroup createHostnameResolver(final String host) {
+ return DnsAddressEndpointGroup.of(host);
+ }
+
+ /**
+ * Resolves {@code ip} from the stored ServiceEntries. A DNS-resolved ServiceEntry matches only once its hostname
+ * resolution has completed; until then the IP stays unknown.
+ *
+ * @return the composed service of the first matching ServiceEntry by namespace and name, or the unknown service
+ */
+ protected ServiceMetaInfo resolveServiceMetaInfo(final String ip) {
+ final Optional<ServiceEntry> serviceEntry =
+ IstioServiceEntries.INSTANCE.list()
+ .stream()
+ .filter(this::isConsidered)
+ .filter(it -> matches(it.getSpec(), ip))
+ .findFirst();
+
+ if (serviceEntry.isEmpty()) {
+ log.debug("No corresponding service entry for IP: {}", ip);
+ return config.serviceMetaInfoFactory().unknown();
+ }
+
+ log.debug("Composing service meta info from service entry for IP: {}", ip);
+ return composeServiceMetaInfo(serviceEntry.get(), ip);
+ }
+
+ private boolean isConsidered(final ServiceEntry serviceEntry) {
+ return serviceEntry.getMetadata() != null
+ && serviceEntry.getSpec() != null
+ && !ignoredNamespaces.contains(serviceEntry.getMetadata().getNamespace());
+ }
+
+ private static boolean isResolvedByDns(final ServiceEntrySpec spec) {
+ return spec.getResolution() != null && (
+ spec.getResolution() == ServiceEntryResolution.DNS
+ || spec.getResolution() == ServiceEntryResolution.DNS_ROUND_ROBIN);
+ }
+
+ private boolean matches(final ServiceEntrySpec spec, final String ip) {
+ if (spec.getResolution() == null) {
+ log.debug("Unsupported service entry resolution: {}", spec.getResolution());
+ return false;
+ }
+ switch (spec.getResolution()) {
+ case STATIC:
+ return (spec.getAddresses() != null
+ && spec.getAddresses().stream().anyMatch(address -> isAddressMatched(address, ip)))
+ || (spec.getEndpoints() != null
+ && spec.getEndpoints()
+ .stream()
+ .map(WorkloadEntrySpec::getAddress)
+ .anyMatch(address -> Objects.equals(address, ip)));
+ case DNS:
+ case DNS_ROUND_ROBIN:
+ return spec.getHosts() != null
+ && spec.getHosts()
+ .stream()
+ .map(hostnameResolvers::get)
+ .filter(Objects::nonNull)
+ .filter(resolver -> resolver.whenReady().isDone())
+ .anyMatch(resolver -> resolver.endpoints()
+ .stream()
+ .anyMatch(endpoint -> Objects.equals(endpoint.ipAddr(), ip)));
+ default:
+ log.debug("Unsupported service entry resolution: {}", spec.getResolution());
+ return false;
+ }
+ }
+
+ private static boolean isAddressMatched(final String address, final String ip) {
+ if (address.contains("/")) {
+ try {
+ return new SubnetUtils(address).getInfo().isInRange(ip);
+ } catch (final IllegalArgumentException e) {
+ // SubnetUtils supports IPv4 only.
+ return false;
+ }
+ }
+ return Objects.equals(ip, address);
}
Review Comment:
`address.contains("/")` will throw a `NullPointerException` if the `addresses` list contains a null entry. Since `matches()` streams over `spec.getAddresses()` without filtering nulls, a malformed/partial ServiceEntry could crash lookups. Consider guarding against null in `isAddressMatched` (early return false) and/or filtering nulls where addresses are streamed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org