suvodeep-pyne commented on code in PR #20067: URL: https://github.com/apache/pinot/pull/20067#discussion_r4397988867
########## pinot-spi/src/main/java/org/apache/pinot/spi/ingestion/error/IngestionSourceErrorRules.java: ########## @@ -0,0 +1,758 @@ +/** + * 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.pinot.spi.ingestion.error; + +import java.net.BindException; +import java.net.SocketException; +import java.net.SocketTimeoutException; +import java.net.UnknownHostException; +import java.net.http.HttpTimeoutException; +import java.util.Collections; +import java.util.HashSet; +import java.util.List; +import java.util.Locale; +import java.util.Set; +import java.util.regex.Matcher; +import java.util.regex.Pattern; +import javax.annotation.Nullable; +import javax.net.ssl.SSLException; + + +/// The built-in rules [IngestionSourceErrorClassifier] applies, organised by SDK. +/// +/// Rules key on the source's own class and code first, and on an HTTP status only where the source has no code of +/// its own (a schema registry, Pulsar's admin API, an OAuth token endpoint) or AWS returned none. They read a message +/// only where the client stringified its evidence or the class covers both our side and the source's; each such match +/// is commented at its rule and pinned by a fixture test against the SDK version Pinot ships. A phrase an SDK upgrade +/// changes leaves the failure unclassified, never in the wrong category. +/// +/// They never key on Pinot's own wrappers, an outer plugin message, a bare Kafka `SerializationException` or +/// `InvalidConfigurationException`, or which exceptions a client considers retriable. +/// +/// A rule returns a verdict to decide its node, [IngestionSourceErrorVerdict#none] when the failure is not the +/// source's verdict on our request, or null to leave the node undecided. +final class IngestionSourceErrorRules { + private IngestionSourceErrorRules() { + } + + private static final String TARGET_ADMIN = IngestionSourceErrorVerdict.TARGET_ADMIN; + private static final String TARGET_BROKER = IngestionSourceErrorVerdict.TARGET_BROKER; + private static final String TARGET_IDP = IngestionSourceErrorVerdict.TARGET_IDP; + private static final String TARGET_KMS = IngestionSourceErrorVerdict.TARGET_KMS; + private static final String TARGET_SCHEMA_REGISTRY = IngestionSourceErrorVerdict.TARGET_SCHEMA_REGISTRY; + private static final String TARGET_STS = IngestionSourceErrorVerdict.TARGET_STS; + + // Our own interrupts and cancellations, matched exactly: SocketTimeoutException extends InterruptedIOException. + private static final Set<String> INTERRUPTS = Set.of("java.lang.InterruptedException", + "java.io.InterruptedIOException", "java.nio.channels.ClosedByInterruptException", + "java.util.concurrent.CancellationException"); + // Metadata endpoints serve the process's own ambient identity: AWS and Azure IMDS, the ECS and EKS Pod Identity + // container endpoints, the GCE metadata server. + private static final List<String> METADATA_ENDPOINTS = List.of("169.254.169.254", "169.254.170.2", + "169.254.170.23", "metadata.google.internal"); + // Socket failures on our side before any request: ephemeral ports or file descriptors exhausted. + private static final String OUR_SOCKETS_EXHAUSTED = "Too many open files"; + + // --------------------------------------------------------------------------------------------------------------- + // Evidence recorded by a client that is about to discard it + + @Nullable + static IngestionSourceErrorVerdict matchEvidence(IngestionSourceErrorEvidence evidence) { + String code = evidence.getCode(); + if (IngestionSourceErrorEvidence.UNREACHABLE.equals(code)) { + return sourceRead(code, evidence.getTarget()); + } + if (IngestionSourceErrorEvidence.LOCAL_PRECHECK.equals(code)) { + return auth(IngestionSourceErrorSubReason.AUTHENTICATION, code, evidence.getTarget()); + } + if (IngestionSourceErrorEvidence.LOCAL_SETUP.equals(code)) { + return none(code); + } + if (IngestionSourceErrorEvidence.UNKNOWN_TOPIC_OR_PARTITION.equals(code)) { + return sourceRead(code, evidence.getTarget() != null ? evidence.getTarget() : TARGET_BROKER); + } + // A status or another code means something only to the source that recorded it: a registered rule reads it. + return null; + } + + /// Source-specific rules for one node, reading the part of its chain beneath it where a rule needs to. + @Nullable + static IngestionSourceErrorVerdict matchSource(Throwable node, List<Throwable> below, + @Nullable IngestionSourceErrorEvidence evidence) { + Class<?> clazz = node.getClass(); + if (INTERRUPTS.contains(clazz.getName())) { + return none(clazz.getSimpleName()); + } + if (hasClass(clazz, AWS_SERVICE_EXCEPTION)) { + return awsService(node, evidence); + } + if (hasClass(clazz, AWS_CLIENT_EXCEPTION)) { + return awsClient(node, below); + } + if (hasClass(clazz, CONFLUENT_REST_CLIENT_EXCEPTION)) { + return schemaRegistry(node); + } + IngestionSourceErrorVerdict kafka = kafka(node, below, evidence); + if (kafka != null) { + return kafka; + } + if (hasClass(clazz, LOGIN_EXCEPTION)) { + return login(node); + } + if (hasClass(clazz, PULSAR_CLIENT_EXCEPTION) || hasClass(clazz, PULSAR_V5_CLIENT_EXCEPTION)) { + return pulsar(node); + } + if (hasClass(clazz, PULSAR_ADMIN_EXCEPTION)) { + return pulsarAdmin(node); + } + if (hasClass(clazz, GAX_API_EXCEPTION)) { + Object statusCode = invoke(node, "getStatusCode"); + return grpc(statusCode != null ? invoke(statusCode, "getCode") : null, node, evidence); + } + if (hasClass(clazz, GRPC_STATUS_RUNTIME_EXCEPTION) || hasClass(clazz, GRPC_STATUS_EXCEPTION)) { + Object status = invoke(node, "getStatus"); + return grpc(status != null ? invoke(status, "getCode") : null, node, evidence); + } + if (hasClass(clazz, GOOGLE_AUTH_EXCEPTION)) { + return googleAuth(node, below, evidence); + } + return null; + } + + /// Generic transport rules, applied only when no source-specific rule decided the chain. + @Nullable + static IngestionSourceErrorVerdict matchGeneric(Throwable node) { + return isUnreachable(node) ? unreachable(node, "") : null; + } + + /// Whether `throwable` means the endpoint never answered: DNS, connect, TLS or a socket timeout. + private static boolean isUnreachable(Throwable throwable) { + if (throwable instanceof BindException + || (throwable instanceof SocketException && messageOf(throwable).contains(OUR_SOCKETS_EXHAUSTED))) { + return false; + } + return throwable instanceof UnknownHostException || throwable instanceof SocketException + || throwable instanceof SocketTimeoutException || throwable instanceof HttpTimeoutException + || throwable instanceof SSLException; Review Comment: Fixed. Before, only the Kafka broker rule read a peer's TLS alert. These now read it too: - the generic rule; - the AWS SDK client rule; - gRPC `UNAVAILABLE`, which is how gRPC surfaces a refused handshake (the SSL failure is its cause). A certificate rejection (`bad_certificate`, `unknown_ca`, `certificate_required`, ...) is AUTH/authentication, and `access_denied` is AUTH/authorization. Any other handshake failure stays SOURCE_READ. Your probe now gives `AUTH code=bad_certificate`. Tests: `peerRejectingOurCertificateIsAuthOverAnyTransport` in pinot-spi, plus a gRPC row in the GCS fixtures and an AWS row in the Kinesis fixtures. ########## pinot-spi/src/main/java/org/apache/pinot/spi/ingestion/error/IngestionSourceErrorClassifier.java: ########## @@ -0,0 +1,295 @@ +/** + * 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.pinot.spi.ingestion.error; + +import java.util.ArrayDeque; +import java.util.ArrayList; +import java.util.Collections; +import java.util.Comparator; +import java.util.Deque; +import java.util.IdentityHashMap; +import java.util.List; +import java.util.Optional; +import java.util.Set; +import java.util.concurrent.CopyOnWriteArraySet; +import javax.annotation.Nullable; + + +/// Folds a failure to reach an ingestion source into [IngestionSourceErrorCategory#AUTH] or +/// [IngestionSourceErrorCategory#SOURCE_READ], from the source's own verdict on our request. Status codes and error +/// codes are only evidence for that verdict, read per source. +/// +/// The exception tree is split into chains: the failure with its causes, and each suppressed exception with its own +/// causes. When an exception in the tree carries boundary evidence ([IngestionSourceErrorEvidence#setBoundary]), only +/// the parts of chains at or beneath a boundary are read; otherwise the caller vouches that the whole failure came +/// from the source. Within a chain: +/// +/// 1. walking from the root cause up, the first source-specific rule that decides wins. A rule that says the failure +/// is not the source's verdict (our clock, our ambient identity, a credential that failed to load) therefore +/// decides only when no source exception sits beneath it, and a wrapper above it can't reinstate a category; +/// 2. generic rules, which read the transport rather than a source's answer (`java.net` / `javax.net.ssl`, and any +/// registered with [#registerGenericRule]), apply only when no source rule decided the chain. +/// +/// Across chains [IngestionSourceErrorCategory#AUTH] beats [IngestionSourceErrorCategory#SOURCE_READ]: a refusal +/// recorded as a suppressed exception still surfaces. +/// +/// Built-in rules cover kafka-clients, the Confluent schema-registry client, the AWS SDK v2, the Pulsar client and +/// admin APIs, gRPC and gax, Kerberos and OAuth logins, and JDK network types. They match classes by name anywhere in +/// the hierarchy, because SDKs load in plugin classloaders, sometimes in two versions at once. They read codes +/// reflectively and fail closed. A plugin for another SDK adds rules with [#registerRule]. +/// +/// [#classify] is pure and never throws, apart from rethrowing the JVM's own fatal errors (`OutOfMemoryError`, +/// `InternalError`). +public final class IngestionSourceErrorClassifier { + // Bounds the walk over pathological exception graphs; real failures are a few dozen nodes. + private static final int MAX_NODES = 1_000; + private static final Set<IngestionSourceErrorRule> REGISTERED_RULES = new CopyOnWriteArraySet<>(); + private static final Set<IngestionSourceErrorRule> REGISTERED_GENERIC_RULES = new CopyOnWriteArraySet<>(); + + private IngestionSourceErrorClassifier() { + } + + /// Classifies `failure`, or returns empty when it is not the source's verdict on our request. + public static Optional<IngestionSourceErrorClassification> classify(Throwable failure) { + try { + Walk walk = new Walk(); + walk.visit(failure); + walk.decide(); + Decision decided = null; + for (Decision decision : walk._decisions) { Review Comment: Left to callers, by design. A decision is per chain because a suppressed exception can be independent of the main one: for example, a client that attaches the per-server errors it saw under its own timeout. - **Batch callers:** a caller that mixes a source call with other work scopes the read with `setBoundary`. A close failure suppressed onto the source call does not inherit the boundary (`boundaryLimitsTheReadToTheSourcesBranches`). - **Realtime sites (part 2):** they record SOURCE_READ only when Pinot gives up on the source. The transient fetch handler skips recording when `_shouldStop` is set, including when cancellation becomes the sixth failure. ########## pinot-spi/src/main/java/org/apache/pinot/spi/ingestion/error/IngestionSourceErrorClassifier.java: ########## @@ -0,0 +1,295 @@ +/** + * 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.pinot.spi.ingestion.error; + +import java.util.ArrayDeque; +import java.util.ArrayList; +import java.util.Collections; +import java.util.Comparator; +import java.util.Deque; +import java.util.IdentityHashMap; +import java.util.List; +import java.util.Optional; +import java.util.Set; +import java.util.concurrent.CopyOnWriteArraySet; +import javax.annotation.Nullable; + + +/// Folds a failure to reach an ingestion source into [IngestionSourceErrorCategory#AUTH] or +/// [IngestionSourceErrorCategory#SOURCE_READ], from the source's own verdict on our request. Status codes and error +/// codes are only evidence for that verdict, read per source. +/// +/// The exception tree is split into chains: the failure with its causes, and each suppressed exception with its own +/// causes. When an exception in the tree carries boundary evidence ([IngestionSourceErrorEvidence#setBoundary]), only +/// the parts of chains at or beneath a boundary are read; otherwise the caller vouches that the whole failure came +/// from the source. Within a chain: +/// +/// 1. walking from the root cause up, the first source-specific rule that decides wins. A rule that says the failure +/// is not the source's verdict (our clock, our ambient identity, a credential that failed to load) therefore +/// decides only when no source exception sits beneath it, and a wrapper above it can't reinstate a category; +/// 2. generic rules, which read the transport rather than a source's answer (`java.net` / `javax.net.ssl`, and any +/// registered with [#registerGenericRule]), apply only when no source rule decided the chain. +/// +/// Across chains [IngestionSourceErrorCategory#AUTH] beats [IngestionSourceErrorCategory#SOURCE_READ]: a refusal +/// recorded as a suppressed exception still surfaces. +/// +/// Built-in rules cover kafka-clients, the Confluent schema-registry client, the AWS SDK v2, the Pulsar client and +/// admin APIs, gRPC and gax, Kerberos and OAuth logins, and JDK network types. They match classes by name anywhere in +/// the hierarchy, because SDKs load in plugin classloaders, sometimes in two versions at once. They read codes +/// reflectively and fail closed. A plugin for another SDK adds rules with [#registerRule]. +/// +/// [#classify] is pure and never throws, apart from rethrowing the JVM's own fatal errors (`OutOfMemoryError`, +/// `InternalError`). +public final class IngestionSourceErrorClassifier { + // Bounds the walk over pathological exception graphs; real failures are a few dozen nodes. + private static final int MAX_NODES = 1_000; + private static final Set<IngestionSourceErrorRule> REGISTERED_RULES = new CopyOnWriteArraySet<>(); + private static final Set<IngestionSourceErrorRule> REGISTERED_GENERIC_RULES = new CopyOnWriteArraySet<>(); + + private IngestionSourceErrorClassifier() { + } + + /// Classifies `failure`, or returns empty when it is not the source's verdict on our request. + public static Optional<IngestionSourceErrorClassification> classify(Throwable failure) { + try { + Walk walk = new Walk(); + walk.visit(failure); + walk.decide(); + Decision decided = null; + for (Decision decision : walk._decisions) { + if (decision._verdict.getCategory() == IngestionSourceErrorCategory.AUTH) { + decided = decision; + break; + } + if (decided == null && decision._verdict.getCategory() == IngestionSourceErrorCategory.SOURCE_READ) { + decided = decision; + } + } + if (decided == null) { + return Optional.empty(); + } + return Optional.of(new IngestionSourceErrorClassification(decided._verdict, decided._node, decided._evidence)); + } catch (Throwable t) { + // Observation only: a classification bug, or a hostile exception, must never change how the failure itself is + // handled. + rethrowIfFatal(t); + return Optional.empty(); + } + } + + /// Running out of memory or a broken JVM is not ours to swallow; a stack overflow from a hostile exception is. + static void rethrowIfFatal(Throwable t) { + if (t instanceof VirtualMachineError && !(t instanceof StackOverflowError)) { + throw (VirtualMachineError) t; + } + } + + /// Adds rules for an SDK Pinot doesn't ship. Registering the same rule twice has no effect, and the order rules are + /// registered in never changes an outcome. + public static void registerRule(IngestionSourceErrorRule rule) { Review Comment: Agreed, this needs an owner decision. The reasons for `pinot-spi`, and the options: - **Why `pinot-spi`:** stream plugins and decoders load in their own classloaders, and both need the evidence carrier and the rule registry. `pinot-spi` is the only module that the server, the controller and every plugin share. - **Public surface:** the rule table, `IngestionSourceErrorRules`, is package-private. The public types are: - the classifier and its two registries; - the rule interface with its reflection helpers; - the evidence carrier; - the classification and verdict values and their three enums. - **Message text:** a rule reads text only where the client put its evidence there, and each phrase is pinned by a fixture. A phrase that an SDK upgrade changes can move a failure to another category, so those fixtures are part of checking any upgrade of these SDKs. - **Option 1, review with the first consumer:** part 2 is ready: the server and controller call sites, and the gauges. I can open it stacked on this PR. - **Option 2, a smaller SPI:** each plugin could register its own SDK rules through the registry, which moves the SDK-specific matching out of `pinot-spi`. Today built-in rules win over registered ones on the same exception, so this would need care to keep that precedence, the registration lifecycle and the classloader behaviour, with equivalence tests. -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
