diff --git a/docs/src/deployment/observability.md b/docs/src/deployment/observability.md index e0d179e4..aa71a4e3 100644 --- a/docs/src/deployment/observability.md +++ b/docs/src/deployment/observability.md @@ -28,48 +28,83 @@ The app exposes health endpoints and uses OTLP export for traces and metrics whe ## Metrics -### Metrics exported by Spring Boot and Micrometer - -The application exposes the standard Micrometer metrics and OTLP export can be enabled through environment variables. - -```yaml -management: - tracing: - sampling: - probability: ${OTEL_TRACES_SAMPLER_ARG:0.1} - export: - enabled: ${OTEL_TRACING_ENABLED:false} - - otlp: - metrics: - export: - url: ${OTEL_EXPORTER_OTLP_METRICS_ENDPOINT:${OTEL_EXPORTER_OTLP_ENDPOINT:http://localhost:4318}/v1/metrics} - enabled: ${OTEL_METRICS_EXPORT_ENABLED:false} - transport: ${OTEL_EXPORTER_OTLP_METRICS_PROTOCOL:${OTEL_EXPORTER_OTLP_PROTOCOL:http/protobuf}} - - opentelemetry: - tracing: - export: - otlp: - endpoint: ${OTEL_EXPORTER_OTLP_TRACES_ENDPOINT:${OTEL_EXPORTER_OTLP_ENDPOINT:http://localhost:4318}/v1/traces} - transport: ${OTEL_EXPORTER_OTLP_TRACES_TRANSPORT:${OTEL_EXPORTER_OTLP_PROTOCOL:http}} - resource-attributes: - deployment.environment: ${OTEL_ENV:local} - service.namespace: ${OTEL_SERVICE_NAMESPACE:idp-core} - service.name: ${OTEL_SERVICE_NAME:internal-developer-platform} +### Native OpenTelemetry metrics + +The application uses `io.opentelemetry.instrumentation:opentelemetry-spring-boot-starter` to configure the SDK, metric readers, and OTLP exporters. You do not need application-defined `OpenTelemetry` or `SdkMeterProvider` beans. The ingestion processors use the auto-configured instance. + +```yaml title="application.yml" +otel: + metrics: + exporter: ${OTEL_METRICS_EXPORTER:otlp} + exporter: + otlp: + endpoint: ${OTEL_EXPORTER_OTLP_ENDPOINT:http://localhost:4318} + protocol: ${OTEL_EXPORTER_OTLP_PROTOCOL:http/protobuf} ``` -### Custom business metrics +The previous Spring Boot starter exported Micrometer metrics. Its SDK configuration could attach an existing meter provider, but did not create one for these native instruments. Setting `management.opentelemetry.metrics.export.otlp.*` did not activate collection. A `DefaultMeter` therefore discarded custom measurements even while traces worked. -Business-level metrics are intentionally separate from Camel route telemetry. The application adds custom counters only for functional decisions that Camel does not expose as a reusable business signal. +`camel-opentelemetry2-starter` provides Camel tracing, not automatic base route metrics. This correction restores the existing custom metrics; it does not add Camel base metrics or export Micrometer registries. -Examples: +### Diagnose missing metrics + +1. Restart the application and inspect `[OTEL SDK CHECK]`. `DefaultMeter` means no-op collection; `SdkMeter` confirms a native SDK meter, not successful delivery to Datadog. +2. Temporarily set `OTEL_METRICS_EXPORTER=console,otlp` and `OTEL_METRIC_EXPORT_INTERVAL=5000`. The starter includes the logging exporter. Console and OTLP exporters can coexist without a manually created reader. +3. Send a webhook, then wait for the export interval. Look for `idp.webhook.security.requests.total` or `idp.webhook.mapping.applied.total`. A filtered mapping still records attempts and filtered results, but does not create an entity. +4. If console metrics appear but Datadog receives nothing, check exporter errors, collector metrics routing, authentication, and Datadog filters. Configure required request headers with `OTEL_EXPORTER_OTLP_METRICS_HEADERS`; resource attributes are not HTTP headers. +5. Restore `OTEL_METRICS_EXPORTER=otlp` and the normal interval after diagnosis. Console metrics can contain operational identifiers and generate large logs. + +> [!WARNING] +> On platforms that allocate CPU only during requests, background export can pause between requests. Check CPU allocation and instance lifetime if collection works but periodic export is delayed. + +For HTTP/protobuf, set `OTEL_EXPORTER_OTLP_METRICS_ENDPOINT` to the full collector URL, including `/v1/metrics`. Inside a container, `localhost` points to the application container. Use the collector service name, or `host.docker.internal` if the collector runs on your macOS host through Docker Desktop. The repository's Docker Compose file does not start a collector. + +### Verify the export path -- `idp.webhook.security.requests.total{connector, security_type}` -- `idp.webhook.security.result{connector, security_type, result}` -- `idp.webhook.security.validation.errors.total{connector, security_type, reason}` +Run the focused tests without a collector or Datadog credentials: -These metrics reflect the real security decision made by the application and should not duplicate the standard Camel `route.*` metrics. +```bash +mvn -Dspring.profiles.active=test -Dtest=OpenTelemetryMetricsAutoConfigurationTest test +``` + +The tests distinguish these checks: + +- **SDK collection**: an in-memory exporter receives a custom counter and histogram with the expected measurements. A separate test checks the warning when you disable the SDK. +- **OTLP HTTP transport**: the real exporter sends a `protobuf` request to a local server on an ephemeral port. The test checks the method, path, content type, configured header, and encoded metric, connector, and service identifiers. An HTTP 200 response completes the export; an HTTP 400 response reports failure. This check does not decode numeric points or exercise periodic scheduling. +- **Datadog delivery**: these tests do not contact your deployed collector or Datadog. After deployment, send a webhook and confirm reception at the collector, then in Datadog. Check that the collector enables an OTLP receiver and a metrics pipeline connected to the Datadog exporter, or that your Datadog Agent enables OTLP metrics ingestion. A successful HTTP response alone does not prove downstream delivery. + +> [!WARNING] +> Do not use `SdkMeterProvider.forceFlush().isSuccess()` as proof of HTTP acceptance. In this SDK version, the reader can complete the flush successfully even when the exporter receives HTTP 400. The transport tests observe the actual exporter result; in deployment, check exporter errors and collector reception. + +### Custom webhook ingestion metrics + +The application emits custom OpenTelemetry metrics in the webhook ingestion pipeline to capture business decisions and failures that are not exposed by the standard Camel route telemetry. + +The following metrics are produced by the ingestion, security and decoding processors: + +| Metric name | Attributes | Meaning | +|--------------------------------------------------|-----------------------------------------|------------------------------------------------------------------------------| +| `idp.webhook.security.requests.total` | `connector`, `security_type` | Number of security checks executed for a connector. | +| `idp.webhook.security.result` | `connector`, `security_type`, `result` | Result of the security decision (`allowed`, `rejected`, `error`). | +| `idp.webhook.security.validation.errors.total` | `connector`, `security_type`, `reason` | Security validation failures and malformed configuration errors. | +| `idp.webhook.payload.decoded.total` | `connector`, `encoding` | Successful payload decodes for each encoding step. | +| `idp.webhook.payload.decode.errors.total` | `connector`, `reason` | Payload decode failures, including unsupported or corrupted encodings. | +| `idp.webhook.payload.size.bytes` | `connector` | Histogram of decoded payload size in bytes. | +| `idp.webhook.mapping.applied.total` | `connector`, `mapping_id`, `action` | Mapping execution attempts. | +| `idp.webhook.mapping.filtered.total` | `connector`, `mapping_id`, `reason` | Mapping results rejected by the filter stage. | +| `idp.webhook.mapping.result` | `connector`, `mapping_id`, `result` | Mapping outcome (`MATCHED`, `FILTERED_OUT`, `ERROR`). | +| `idp.webhook.mapping.errors.total` | `connector`, `mapping_id`, `error_type` | Mapping exceptions and runtime failures. | +| `idp.webhook.mapping.latency.ms` | `connector`, `mapping_id` | Mapping duration histogram in milliseconds. | +| `idp.webhook.entity.created.total` | `template`, `connector` | Number of entities created during ingestion. | +| `idp.webhook.entity.updated.total` | `template`, `connector` | Number of entities updated during ingestion. | +| `idp.webhook.entity.deleted.total` | `template`, `connector` | Number of entities deleted during ingestion. | +| `idp.webhook.entity.patch.total` | `template`, `connector`, `action` | Number of patch operations applied. | +| `idp.webhook.entity.validation.errors.total` | `template`, `connector`, `reason` | Entity validation and business-rule failures. | +| `idp.webhook.entity.notfound.total` | `template`, `connector`, `action` | Missing entity operations during ingestion. | +| `idp.webhook.entity.conflict.total` | `template`, `connector`, `reason` | Conflict scenarios such as duplicate entities. | +| `idp.webhook.functional.validation.failed.total` | `connector`, `template`, `reason` | Functional validation failures captured after a mapping or entity operation. | + +These metric names are centralized in `IngestionObservabilityConstants` and are intentionally aligned with the connector, template, action, and reason dimensions used across the webhook processing pipeline. ## Logging @@ -85,19 +120,23 @@ OpenTelemetry tracing is enabled through Spring Boot configuration. The applicat | Variable | Description | Default | | --- | --- | --- | -| `OTEL_TRACING_ENABLED` | Enables tracing export | `false` | -| `OTEL_METRICS_EXPORT_ENABLED` | Enables OTLP metrics export | `false` | +| `OTEL_TRACES_EXPORTER` | Trace exporter; `none` disables export | `otlp` (`none` locally) | +| `OTEL_METRICS_EXPORTER` | Metric exporters; `console,otlp` for diagnosis | `otlp` | +| `OTEL_METRIC_EXPORT_INTERVAL` | Metric export interval in milliseconds | `60000` | +| `OTEL_LOGS_EXPORTER` | Log exporter | `none` | | `OTEL_TRACES_SAMPLER_ARG` | Trace sampling ratio | `0.1` | | `OTEL_EXPORTER_OTLP_ENDPOINT` | Base OTLP endpoint | `http://localhost:4318` | | `OTEL_EXPORTER_OTLP_TRACES_ENDPOINT` | Full traces endpoint | `/v1/traces` | | `OTEL_EXPORTER_OTLP_METRICS_ENDPOINT` | Full metrics endpoint | `/v1/metrics` | -| `OTEL_EXPORTER_OTLP_PROTOCOL` | Generic OTLP protocol fallback | `http/protobuf` for metrics, `http` for traces | -| `OTEL_EXPORTER_OTLP_TRACES_TRANSPORT` | Trace-specific transport override | `OTEL_EXPORTER_OTLP_PROTOCOL` or `http` | +| `OTEL_EXPORTER_OTLP_PROTOCOL` | OTLP protocol, `http/protobuf` or `grpc` | `http/protobuf` | +| `OTEL_EXPORTER_OTLP_TRACES_PROTOCOL` | Trace-specific protocol override | `OTEL_EXPORTER_OTLP_PROTOCOL` | | `OTEL_EXPORTER_OTLP_METRICS_PROTOCOL` | Metrics-specific transport override | `OTEL_EXPORTER_OTLP_PROTOCOL` | | `OTEL_RESOURCE_ATTRIBUTES` | Resource metadata injected by the environment | - | The signal-specific variables win over the generic fallback, so local development and hosted deployments can both work without hard-coded vendor configuration. +Replace the old `OTEL_TRACING_ENABLED` and `OTEL_METRICS_EXPORT_ENABLED` flags with exporter selection (`otlp` or `none`). Replace `*_TRANSPORT=http` with `*_PROTOCOL=http/protobuf`. The official starter reads `otel.*`, not the former `management.opentelemetry.*` export settings. The test profile disables network export for all signals. + ### Important trace attributes The application sets trace attributes for the webhook pipeline, including the connector identifier and the security decision. @@ -119,7 +158,9 @@ The app exposes health endpoints used by Docker and Kubernetes. | `/actuator/health/liveness` | Liveness probe | | `/actuator/health/readiness` | Readiness probe | -## Next steps +--- + +## Next Steps - **[Configuration](configuration.md)** - Environment and Spring Boot settings - **[Docker](docker.md)** - Docker deployment and OTLP examples diff --git a/pom.xml b/pom.xml index beeaf5eb..538c508a 100644 --- a/pom.xml +++ b/pom.xml @@ -39,6 +39,14 @@ + + + io.opentelemetry.instrumentation + opentelemetry-instrumentation-bom + 2.32.0 + pom + import + io.netty netty-bom @@ -180,6 +188,12 @@ test + + io.opentelemetry + opentelemetry-sdk-testing + test + + org.mock-server mockserver-client-java @@ -332,9 +346,10 @@ camel-rest-starter + - org.springframework.boot - spring-boot-starter-opentelemetry + io.opentelemetry.instrumentation + opentelemetry-spring-boot-starter @@ -343,8 +358,8 @@ - org.apache.camel.springboot - camel-micrometer-starter + io.opentelemetry + opentelemetry-api diff --git a/src/main/java/com/decathlon/idp_core/infrastructure/adapters/ingestion/configuration/CamelTracingConfiguration.java b/src/main/java/com/decathlon/idp_core/infrastructure/adapters/ingestion/configuration/CamelTracingConfiguration.java new file mode 100644 index 00000000..65295fd6 --- /dev/null +++ b/src/main/java/com/decathlon/idp_core/infrastructure/adapters/ingestion/configuration/CamelTracingConfiguration.java @@ -0,0 +1,49 @@ +package com.decathlon.idp_core.infrastructure.adapters.ingestion.configuration; + +import org.apache.camel.CamelContext; +import org.apache.camel.spring.boot.CamelContextConfiguration; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + +import io.opentelemetry.api.OpenTelemetry; +import io.opentelemetry.api.trace.Tracer; +import io.opentelemetry.context.propagation.ContextPropagators; +import lombok.extern.slf4j.Slf4j; + +@Configuration +@Slf4j +public class CamelTracingConfiguration { + + static final String CAMEL_INSTRUMENTATION_SCOPE = "camel"; + + @Bean + public Tracer camelTracer(OpenTelemetry openTelemetry) { + return openTelemetry.getTracer(CAMEL_INSTRUMENTATION_SCOPE); + } + + @Bean + public ContextPropagators camelContextPropagators(OpenTelemetry openTelemetry) { + return openTelemetry.getPropagators(); + } + + @Bean + public CamelContextConfiguration camelTracingBridge(Tracer camelTracer, + ContextPropagators camelContextPropagators) { + return new CamelContextConfiguration() { + @Override + public void beforeApplicationStart(CamelContext camelContext) { + // Expose Spring SDK beans directly in Camel registry before route startup. + camelContext.getRegistry().bind("camelTracer", camelTracer); + camelContext.getRegistry().bind("camelContextPropagators", camelContextPropagators); + log.info( + "[OTEL CAMEL BRIDGE] Bound Tracer={} and ContextPropagators={} into Camel registry", + camelTracer.getClass().getName(), camelContextPropagators.getClass().getName()); + } + + @Override + public void afterApplicationStart(CamelContext camelContext) { + // No-op + } + }; + } +} diff --git a/src/main/java/com/decathlon/idp_core/infrastructure/adapters/ingestion/configuration/IngestionMetricsConfiguration.java b/src/main/java/com/decathlon/idp_core/infrastructure/adapters/ingestion/configuration/IngestionMetricsConfiguration.java new file mode 100644 index 00000000..2c7f5ab2 --- /dev/null +++ b/src/main/java/com/decathlon/idp_core/infrastructure/adapters/ingestion/configuration/IngestionMetricsConfiguration.java @@ -0,0 +1,26 @@ +package com.decathlon.idp_core.infrastructure.adapters.ingestion.configuration; + +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + +import io.opentelemetry.api.OpenTelemetry; +import io.opentelemetry.api.metrics.Meter; + +@Configuration +public class IngestionMetricsConfiguration { + + @Bean("ingestionMeter") + public Meter ingestionMeter(OpenTelemetry openTelemetry) { + return openTelemetry.getMeter("com.decathlon.idp_core.ingestion"); + } + + @Bean("securityMeter") + public Meter securityMeter(OpenTelemetry openTelemetry) { + return openTelemetry.getMeter("com.decathlon.idp_core.security"); + } + + @Bean("decodingMeter") + public Meter decodingMeter(OpenTelemetry openTelemetry) { + return openTelemetry.getMeter("com.decathlon.idp_core.decoding"); + } +} diff --git a/src/main/java/com/decathlon/idp_core/infrastructure/adapters/ingestion/configuration/IngestionObservabilityConstants.java b/src/main/java/com/decathlon/idp_core/infrastructure/adapters/ingestion/configuration/IngestionObservabilityConstants.java new file mode 100644 index 00000000..b245664d --- /dev/null +++ b/src/main/java/com/decathlon/idp_core/infrastructure/adapters/ingestion/configuration/IngestionObservabilityConstants.java @@ -0,0 +1,41 @@ +package com.decathlon.idp_core.infrastructure.adapters.ingestion.configuration; + +public final class IngestionObservabilityConstants { + + public static final String SECURITY_REQUESTS_TOTAL = "idp.webhook.security.requests.total"; + public static final String SECURITY_RESULT = "idp.webhook.security.result"; + public static final String SECURITY_VALIDATION_ERRORS_TOTAL = "idp.webhook.security.validation.errors.total"; + + public static final String PAYLOAD_DECODED_TOTAL = "idp.webhook.payload.decoded.total"; + public static final String PAYLOAD_DECODE_ERRORS_TOTAL = "idp.webhook.payload.decode.errors.total"; + public static final String PAYLOAD_SIZE_BYTES = "idp.webhook.payload.size.bytes"; + + public static final String MAPPING_APPLIED_TOTAL = "idp.webhook.mapping.applied.total"; + public static final String MAPPING_FILTERED_TOTAL = "idp.webhook.mapping.filtered.total"; + public static final String MAPPING_RESULT = "idp.webhook.mapping.result"; + public static final String MAPPING_ERRORS_TOTAL = "idp.webhook.mapping.errors.total"; + public static final String MAPPING_LATENCY_MS = "idp.webhook.mapping.latency.ms"; + + public static final String ENTITY_CREATED_TOTAL = "idp.webhook.entity.created.total"; + public static final String ENTITY_UPDATED_TOTAL = "idp.webhook.entity.updated.total"; + public static final String ENTITY_DELETED_TOTAL = "idp.webhook.entity.deleted.total"; + public static final String ENTITY_PATCH_TOTAL = "idp.webhook.entity.patch.total"; + public static final String ENTITY_VALIDATION_ERRORS_TOTAL = "idp.webhook.entity.validation.errors.total"; + + public static final String FUNCTIONAL_VALIDATION_FAILED_TOTAL = "idp.webhook.functional.validation.failed.total"; + public static final String ENTITY_NOT_FOUND_TOTAL = "idp.webhook.entity.notfound.total"; + public static final String ENTITY_CONFLICT_TOTAL = "idp.webhook.entity.conflict.total"; + + public static final String ATTRIBUTE_CONNECTOR = "connector"; + public static final String ATTRIBUTE_MAPPING_ID = "mapping_id"; + public static final String ATTRIBUTE_ACTION = "action"; + public static final String ATTRIBUTE_REASON = "reason"; + public static final String ATTRIBUTE_TEMPLATE = "template"; + public static final String ATTRIBUTE_RESULT = "result"; + public static final String ATTRIBUTE_SECURITY_TYPE = "security_type"; + + public static final String SPAN_ATTRIBUTE_WEBHOOK_IDENTIFIER = "idp.webhook.identifier"; + + private IngestionObservabilityConstants() { + } +} diff --git a/src/main/java/com/decathlon/idp_core/infrastructure/adapters/ingestion/configuration/OtelLoggingDebugConfig.java b/src/main/java/com/decathlon/idp_core/infrastructure/adapters/ingestion/configuration/OtelLoggingDebugConfig.java new file mode 100644 index 00000000..bd9f0539 --- /dev/null +++ b/src/main/java/com/decathlon/idp_core/infrastructure/adapters/ingestion/configuration/OtelLoggingDebugConfig.java @@ -0,0 +1,41 @@ +package com.decathlon.idp_core.infrastructure.adapters.ingestion.configuration; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.boot.context.event.ApplicationReadyEvent; +import org.springframework.context.event.EventListener; +import org.springframework.stereotype.Component; + +import io.opentelemetry.api.OpenTelemetry; + +/// Inspects the auto-configured telemetry instance without constructing a provider. +@Component +public final class OtelLoggingDebugConfig { + + private static final Logger log = LoggerFactory.getLogger(OtelLoggingDebugConfig.class); + private static final String INSTRUMENTATION_SCOPE = "com.decathlon.idp_core.debug"; + private final OpenTelemetry openTelemetry; + + /// Uses the instance created by the OpenTelemetry starter. + public OtelLoggingDebugConfig(OpenTelemetry openTelemetry) { + this.openTelemetry = openTelemetry; + } + + /// Reports a no-op meter without claiming successful delivery to a backend. + @EventListener(ApplicationReadyEvent.class) + public void debugOtelMeterInitialization() { + var meter = openTelemetry.getMeter(INSTRUMENTATION_SCOPE); + log.info("[OTEL SDK CHECK] OpenTelemetry: {}; MeterProvider: {}; Meter: {}", + openTelemetry.getClass().getName(), openTelemetry.getMeterProvider().getClass().getName(), + meter.getClass().getName()); + // The API fallback is not public; compare runtime types instead of SDK internal + // names. + if (meter.getClass().equals(OpenTelemetry.noop().getMeter(INSTRUMENTATION_SCOPE).getClass())) { + log.warn( + "[OTEL SDK CHECK] No-op Meter: custom metrics are not recorded. Check SDK autoconfiguration and OTEL_SDK_DISABLED."); + } else { + log.info( + "[OTEL SDK CHECK] Meter is not the API no-op. Exporter delivery must still be verified."); + } + } +} diff --git a/src/main/java/com/decathlon/idp_core/infrastructure/adapters/ingestion/metrics/IngestionObservabilitySupport.java b/src/main/java/com/decathlon/idp_core/infrastructure/adapters/ingestion/metrics/IngestionObservabilitySupport.java new file mode 100644 index 00000000..380df142 --- /dev/null +++ b/src/main/java/com/decathlon/idp_core/infrastructure/adapters/ingestion/metrics/IngestionObservabilitySupport.java @@ -0,0 +1,59 @@ +package com.decathlon.idp_core.infrastructure.adapters.ingestion.metrics; + +import static reactor.netty.http.HttpConnectionLiveness.log; + +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + +import io.opentelemetry.api.common.AttributeKey; +import io.opentelemetry.api.common.Attributes; +import io.opentelemetry.api.metrics.LongCounter; +import io.opentelemetry.api.metrics.LongHistogram; +import io.opentelemetry.api.metrics.Meter; +import io.opentelemetry.api.trace.Span; + +public final class IngestionObservabilitySupport { + + private static final Map COUNTER_CACHE = new ConcurrentHashMap<>(); + private static final Map HISTOGRAM_CACHE = new ConcurrentHashMap<>(); + + private IngestionObservabilitySupport() { + } + + public static void incrementCounter(Meter meter, String metricName, String... attributes) { + log.info(">>> [OTEL DEBUG] Increment counter: " + metricName + " with attributes: " + + java.util.Arrays.toString(attributes)); + LongCounter counter = COUNTER_CACHE.computeIfAbsent(metricName, + name -> meter.counterBuilder(name).build()); + counter.add(1, toAttributes(attributes)); + } + + public static void recordHistogram(Meter meter, String metricName, long value, + String... attributes) { + LongHistogram histogram = HISTOGRAM_CACHE.computeIfAbsent(metricName, + name -> meter.histogramBuilder(name).ofLongs().build()); + histogram.record(value, toAttributes(attributes)); + } + + public static void setSpanAttribute(Span span, String attributeName, String attributeValue) { + if (span == null || attributeName == null || attributeName.isBlank() + || attributeValue == null) { + return; + } + span.setAttribute(attributeName, attributeValue); + } + + public static Attributes toAttributes(String... keyValues) { + if (keyValues == null || keyValues.length == 0) { + return Attributes.empty(); + } + if ((keyValues.length % 2) != 0) { + throw new IllegalArgumentException("Metric attributes must be provided as key/value pairs"); + } + var builder = Attributes.builder(); + for (int i = 0; i < keyValues.length; i += 2) { + builder.put(AttributeKey.stringKey(keyValues[i]), keyValues[i + 1]); + } + return builder.build(); + } +} diff --git a/src/main/java/com/decathlon/idp_core/infrastructure/adapters/ingestion/processor/IngestionProcessor.java b/src/main/java/com/decathlon/idp_core/infrastructure/adapters/ingestion/processor/IngestionProcessor.java index 7a88233f..87113993 100644 --- a/src/main/java/com/decathlon/idp_core/infrastructure/adapters/ingestion/processor/IngestionProcessor.java +++ b/src/main/java/com/decathlon/idp_core/infrastructure/adapters/ingestion/processor/IngestionProcessor.java @@ -1,30 +1,42 @@ package com.decathlon.idp_core.infrastructure.adapters.ingestion.processor; +import static com.decathlon.idp_core.infrastructure.adapters.ingestion.configuration.IngestionObservabilityConstants.*; + import java.util.List; +import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.stereotype.Component; +import com.decathlon.idp_core.domain.exception.entity.EntityAlreadyExistsException; import com.decathlon.idp_core.domain.exception.entity.EntityNotFoundException; +import com.decathlon.idp_core.domain.exception.entity.EntityValidationException; import com.decathlon.idp_core.domain.model.entity.Entity; import com.decathlon.idp_core.domain.model.entity.EntityPatch; import com.decathlon.idp_core.domain.model.entity_mapping.EntityDynamicMapping; import com.decathlon.idp_core.domain.model.inbound_connectors.webhook.WebhookConnector; import com.decathlon.idp_core.domain.port.MappingEnginePort; import com.decathlon.idp_core.domain.service.entity.EntityService; +import com.decathlon.idp_core.infrastructure.adapters.ingestion.configuration.IngestionObservabilityConstants; +import com.decathlon.idp_core.infrastructure.adapters.ingestion.metrics.IngestionObservabilitySupport; +import io.opentelemetry.api.metrics.Meter; import io.opentelemetry.api.trace.Span; -import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; @Component @Slf4j -@RequiredArgsConstructor public class IngestionProcessor { - private static final String SPAN_ATTRIBUTE_WEBHOOK_IDENTIFIER = "idp.webhook.identifier"; - private final MappingEnginePort mappingEngine; private final EntityService entityService; + private final Meter meter; + + public IngestionProcessor(MappingEnginePort mappingEngine, EntityService entityService, + @Qualifier("ingestionMeter") Meter meter) { + this.mappingEngine = mappingEngine; + this.entityService = entityService; + this.meter = meter; + } /// Ingests a webhook payload by applying all configured mappings. /// @@ -41,102 +53,187 @@ public class IngestionProcessor { /// @param payload the raw JSON payload from the webhook public void ingest(String payload, WebhookConnector webhookConnectorConfiguration) { String connectorIdentifier = webhookConnectorConfiguration.identifier(); - Span.current().setAttribute(SPAN_ATTRIBUTE_WEBHOOK_IDENTIFIER, connectorIdentifier); - + Span currentSpan = Span.current(); + IngestionObservabilitySupport.setSpanAttribute(currentSpan, SPAN_ATTRIBUTE_WEBHOOK_IDENTIFIER, + connectorIdentifier); + log.info("span id: {}, trace_id: {}", currentSpan.getSpanContext().getSpanId(), + currentSpan.getSpanContext().getTraceId());// todo todelete log.info("Starting ingestion for webhook connector: {}", connectorIdentifier); - webhookConnectorConfiguration.mappings().forEach(mapping -> applyMapping(payload, mapping)); + webhookConnectorConfiguration.mappings() + .forEach(mapping -> applyMapping(payload, mapping, webhookConnectorConfiguration)); log.info("Completed ingestion for webhook connector: {}", connectorIdentifier); } - /// Applies a single mapping to the payload and persists the resulting entity. - /// - /// Skips silently if the mapping filter excludes the payload (returns null). - /// - /// @param payload the raw JSON payload from the webhook - /// @param mapping the mapping definition to apply - private void applyMapping(String payload, EntityDynamicMapping mapping) { - log.debug("Applying mapping for template: {} with action: {}", mapping.identifier(), mapping.action()); - Entity entity = mappingEngine.mapToEntity(payload, mapping); - - if (entity == null) { - log.debug("Mapping filter excluded payload for template: {}", - mapping.entityTemplateIdentifier()); - return; - } - switch (mapping.action()) { - case UPDATE_ENTITY -> handleUpdate(entity); - case UPDATE_PROPERTIES -> handleUpdateProperties(entity); - case UPDATE_RELATIONS -> handleUpdateRelations(entity); - case DELETE_ENTITY -> handleDelete(entity); - case null, default -> log.warn("Unsupported or null mapping action: {}", mapping.action()); + private void applyMapping(String payload, EntityDynamicMapping mapping, + WebhookConnector webhookConnectorConfiguration) { + String connectorIdentifier = webhookConnectorConfiguration.identifier(); + String mappingId = mapping.identifier(); + String action = mapping.action() == null ? "unknown" : mapping.action().name(); + long startTimeNanos = System.nanoTime(); + + incrementCounter(MAPPING_APPLIED_TOTAL, ATTRIBUTE_CONNECTOR, connectorIdentifier, ATTRIBUTE_MAPPING_ID, mappingId, ATTRIBUTE_ACTION, action); + + try { + Entity entity = mappingEngine.mapToEntity(payload, mapping); + + if (entity == null) { + incrementCounter(MAPPING_FILTERED_TOTAL, + ATTRIBUTE_CONNECTOR, connectorIdentifier, ATTRIBUTE_MAPPING_ID, mappingId, ATTRIBUTE_REASON, + "filter_rejected"); + incrementCounter(MAPPING_RESULT, + ATTRIBUTE_CONNECTOR, connectorIdentifier, ATTRIBUTE_MAPPING_ID, mappingId, ATTRIBUTE_RESULT, + "FILTERED_OUT"); + return; + } + + incrementCounter(MAPPING_RESULT, + ATTRIBUTE_CONNECTOR, connectorIdentifier, ATTRIBUTE_MAPPING_ID, mappingId, ATTRIBUTE_RESULT, + "MATCHED"); + + switch (mapping.action()) { + case UPDATE_ENTITY -> handleUpdate(entity, connectorIdentifier); + case UPDATE_PROPERTIES -> handleUpdateProperties(entity, connectorIdentifier); + case UPDATE_RELATIONS -> handleUpdateRelations(entity, connectorIdentifier); + case DELETE_ENTITY -> handleDelete(entity, connectorIdentifier); + case null, default -> log.warn("Unsupported or null mapping action: {}", mapping.action()); + } + + log.info("Successfully processed action {} for entity: {} under template: {}", + mapping.action(), entity.identifier(), entity.templateIdentifier()); + } catch (RuntimeException exception) { + incrementCounter(MAPPING_ERRORS_TOTAL, + ATTRIBUTE_CONNECTOR, connectorIdentifier, ATTRIBUTE_MAPPING_ID, mappingId, + "error_type", exception.getClass().getSimpleName()); + incrementCounter(MAPPING_RESULT, + ATTRIBUTE_CONNECTOR, connectorIdentifier, ATTRIBUTE_MAPPING_ID, mappingId, ATTRIBUTE_RESULT, "ERROR"); + incrementCounter(IngestionObservabilityConstants.FUNCTIONAL_VALIDATION_FAILED_TOTAL, + ATTRIBUTE_CONNECTOR, connectorIdentifier, "mapping_id", mappingId, + ATTRIBUTE_REASON, exception.getClass().getSimpleName()); + recordEntityFailureMetrics(connectorIdentifier, exception); + throw exception; + } finally { + long durationMs = (System.nanoTime() - startTimeNanos) / 1_000_000L; + recordHistogram(IngestionObservabilityConstants.MAPPING_LATENCY_MS, durationMs, + ATTRIBUTE_CONNECTOR, connectorIdentifier, ATTRIBUTE_MAPPING_ID, mappingId); } - - log.info("Successfully processed action {} for entity: {} under template: {}", - mapping.action(), entity.identifier(), entity.templateIdentifier()); } - /// Handles the Update action for an entity. - /// - /// If the entity exists, it is patched with the new data. If the entity does - /// not - /// exist, it is created. - /// - /// @param entity the entity to Update - private void handleUpdate(Entity entity) { - if (entityService.entityExists(entity.templateIdentifier(), entity.identifier())) { - log.debug("Patching entity {} for template: {}", entity.identifier(), - entity.templateIdentifier()); - entityService.patchEntity(entity.templateIdentifier(), entity.identifier(), - EntityPatch.fromEntity(entity)); - } else { - log.debug("Creating entity {} for template: {}", entity.identifier(), - entity.templateIdentifier()); - entityService.createEntity(entity); + private void handleUpdate(Entity entity, String connectorIdentifier) { + try { + if (entityService.entityExists(entity.templateIdentifier(), entity.identifier())) { + log.debug("Patching entity {} for template: {}", entity.identifier(), + entity.templateIdentifier()); + entityService.patchEntity(entity.templateIdentifier(), entity.identifier(), + EntityPatch.fromEntity(entity)); + incrementCounter(IngestionObservabilityConstants.ENTITY_PATCH_TOTAL, ATTRIBUTE_TEMPLATE, + entity.templateIdentifier(), ATTRIBUTE_CONNECTOR, connectorIdentifier, ATTRIBUTE_ACTION, + "UPDATE_ENTITY"); + incrementCounter(IngestionObservabilityConstants.ENTITY_UPDATED_TOTAL, ATTRIBUTE_TEMPLATE, + entity.templateIdentifier(), ATTRIBUTE_CONNECTOR, connectorIdentifier); + } else { + log.debug("Creating entity {} for template: {}", entity.identifier(), + entity.templateIdentifier()); + entityService.createEntity(entity); + incrementCounter(IngestionObservabilityConstants.ENTITY_CREATED_TOTAL, ATTRIBUTE_TEMPLATE, + entity.templateIdentifier(), ATTRIBUTE_CONNECTOR, connectorIdentifier); + } + } catch (RuntimeException exception) { + recordEntityFailureMetrics(connectorIdentifier, exception, entity.templateIdentifier()); + throw exception; } } - /// Handles the Update properties action for an entity. - /// - /// If the entity exists, only its properties are patched. If the entity does - /// not - /// exist, it is created with only its properties. - /// - /// @param entity the entity to Update properties for - private void handleUpdateProperties(Entity entity) { + private void handleUpdateProperties(Entity entity, String connectorIdentifier) { EntityPatch propertiesPatch = new EntityPatch(entity.name(), entity.properties(), List.of()); - if (entityService.entityExists(entity.templateIdentifier(), entity.identifier())) { - entityService.patchEntity(entity.templateIdentifier(), entity.identifier(), propertiesPatch); - } else { - entityService.createEntity(new Entity(null, entity.templateIdentifier(), entity.name(), - entity.identifier(), entity.properties(), List.of())); + try { + if (entityService.entityExists(entity.templateIdentifier(), entity.identifier())) { + entityService.patchEntity(entity.templateIdentifier(), entity.identifier(), + propertiesPatch); + incrementCounter(IngestionObservabilityConstants.ENTITY_PATCH_TOTAL, ATTRIBUTE_TEMPLATE, + entity.templateIdentifier(), ATTRIBUTE_CONNECTOR, connectorIdentifier, ATTRIBUTE_ACTION, + "UPDATE_PROPERTIES"); + incrementCounter(IngestionObservabilityConstants.ENTITY_UPDATED_TOTAL, ATTRIBUTE_TEMPLATE, + entity.templateIdentifier(), ATTRIBUTE_CONNECTOR, connectorIdentifier); + } else { + entityService.createEntity(new Entity(null, entity.templateIdentifier(), entity.name(), + entity.identifier(), entity.properties(), List.of())); + incrementCounter(IngestionObservabilityConstants.ENTITY_CREATED_TOTAL, ATTRIBUTE_TEMPLATE, + entity.templateIdentifier(), ATTRIBUTE_CONNECTOR, connectorIdentifier); + } + } catch (RuntimeException exception) { + recordEntityFailureMetrics(connectorIdentifier, exception, entity.templateIdentifier()); + throw exception; } } - /// Handles the Update relations action for an entity. If the entity exists, - /// only its relations are patched. The entity must exists in order to patch the - /// relations - /// - /// @param entity the entity to Update relations for - /// @throws EntityNotFoundException - private void handleUpdateRelations(Entity entity) { + private void handleUpdateRelations(Entity entity, String connectorIdentifier) { + try { + if (!entityService.entityExists(entity.templateIdentifier(), entity.identifier())) { + throw new EntityNotFoundException(entity.templateIdentifier(), entity.identifier()); + } + EntityPatch relationsPatch = new EntityPatch(null, List.of(), entity.relations()); + + entityService.patchEntity(entity.templateIdentifier(), entity.identifier(), relationsPatch); + incrementCounter(IngestionObservabilityConstants.ENTITY_PATCH_TOTAL, ATTRIBUTE_TEMPLATE, + entity.templateIdentifier(), ATTRIBUTE_CONNECTOR, connectorIdentifier, ATTRIBUTE_ACTION, + "UPDATE_RELATIONS"); + incrementCounter(IngestionObservabilityConstants.ENTITY_UPDATED_TOTAL, ATTRIBUTE_TEMPLATE, + entity.templateIdentifier(), ATTRIBUTE_CONNECTOR, connectorIdentifier); + } catch (RuntimeException exception) { + recordEntityFailureMetrics(connectorIdentifier, exception, entity.templateIdentifier()); + throw exception; + } + } - if (!entityService.entityExists(entity.templateIdentifier(), entity.identifier())) { - throw new EntityNotFoundException(entity.templateIdentifier(), entity.identifier()); + private void handleDelete(Entity entity, String connectorIdentifier) { + try { + entityService.deleteEntity(entity.templateIdentifier(), entity.identifier()); + incrementCounter(IngestionObservabilityConstants.ENTITY_DELETED_TOTAL, ATTRIBUTE_TEMPLATE, + entity.templateIdentifier(), ATTRIBUTE_CONNECTOR, connectorIdentifier); + log.debug("Deleted entity: {} for template: {}", entity.identifier(), + entity.templateIdentifier()); + } catch (RuntimeException exception) { + recordEntityFailureMetrics(connectorIdentifier, exception, entity.templateIdentifier()); + throw exception; } - // Strip properties before invoking entity service - EntityPatch relationsPatch = new EntityPatch(null, List.of(), entity.relations()); + } - entityService.patchEntity(entity.templateIdentifier(), entity.identifier(), relationsPatch); + private void recordEntityFailureMetrics(String connectorIdentifier, RuntimeException exception) { + recordEntityFailureMetrics(connectorIdentifier, exception, "unknown"); + } + private void recordEntityFailureMetrics(String connectorIdentifier, RuntimeException exception, + String templateIdentifier) { + incrementCounter(IngestionObservabilityConstants.ENTITY_VALIDATION_ERRORS_TOTAL, + ATTRIBUTE_TEMPLATE, templateIdentifier, ATTRIBUTE_CONNECTOR, connectorIdentifier, + ATTRIBUTE_REASON, exception.getClass().getSimpleName()); + incrementCounter(IngestionObservabilityConstants.FUNCTIONAL_VALIDATION_FAILED_TOTAL, + ATTRIBUTE_CONNECTOR, connectorIdentifier, ATTRIBUTE_TEMPLATE, templateIdentifier, + ATTRIBUTE_REASON, exception.getClass().getSimpleName()); + + if (exception instanceof EntityNotFoundException) { + incrementCounter(IngestionObservabilityConstants.ENTITY_NOT_FOUND_TOTAL, ATTRIBUTE_TEMPLATE, + templateIdentifier, ATTRIBUTE_CONNECTOR, connectorIdentifier, ATTRIBUTE_ACTION, + "ingestion"); + } + if (exception instanceof EntityAlreadyExistsException) { + incrementCounter(IngestionObservabilityConstants.ENTITY_CONFLICT_TOTAL, ATTRIBUTE_TEMPLATE, + templateIdentifier, ATTRIBUTE_CONNECTOR, connectorIdentifier, ATTRIBUTE_REASON, + "already_exists"); + } + if (exception instanceof EntityValidationException) { + incrementCounter(IngestionObservabilityConstants.ENTITY_VALIDATION_ERRORS_TOTAL, + ATTRIBUTE_TEMPLATE, templateIdentifier, ATTRIBUTE_CONNECTOR, connectorIdentifier, + ATTRIBUTE_REASON, exception.getClass().getSimpleName()); + } } - /// Handles the delete action for an entity. - /// - /// @param entity the entity to delete - private void handleDelete(Entity entity) { - entityService.deleteEntity(entity.templateIdentifier(), entity.identifier()); - log.debug("Deleted entity: {} for template: {}", entity.identifier(), - entity.templateIdentifier()); + private void incrementCounter(String metricName, String... attributes) { + IngestionObservabilitySupport.incrementCounter(meter, metricName, attributes); + } + + private void recordHistogram(String metricName, long value, String... attributes) { + IngestionObservabilitySupport.recordHistogram(meter, metricName, value, attributes); } } diff --git a/src/main/java/com/decathlon/idp_core/infrastructure/adapters/ingestion/processor/SecurityProcessor.java b/src/main/java/com/decathlon/idp_core/infrastructure/adapters/ingestion/processor/SecurityProcessor.java index 104f4ea4..80b9d218 100644 --- a/src/main/java/com/decathlon/idp_core/infrastructure/adapters/ingestion/processor/SecurityProcessor.java +++ b/src/main/java/com/decathlon/idp_core/infrastructure/adapters/ingestion/processor/SecurityProcessor.java @@ -1,17 +1,24 @@ package com.decathlon.idp_core.infrastructure.adapters.ingestion.processor; +import static com.decathlon.idp_core.infrastructure.adapters.ingestion.configuration.IngestionObservabilityConstants.*; + import java.util.List; import java.util.Map; +import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.stereotype.Component; +import com.decathlon.idp_core.domain.exception.webhook.WebhookAuthenticationException; import com.decathlon.idp_core.domain.model.enums.WebhookSecurityType; import com.decathlon.idp_core.domain.model.inbound_connectors.webhook.WebhookConnector; import com.decathlon.idp_core.domain.model.inbound_connectors.webhook.WebhookSecurity; import com.decathlon.idp_core.domain.port.WebhookSecurityStrategy; +import com.decathlon.idp_core.infrastructure.adapters.ingestion.configuration.IngestionObservabilityConstants; import com.decathlon.idp_core.infrastructure.adapters.ingestion.exception.WebhookSecurityException; +import com.decathlon.idp_core.infrastructure.adapters.ingestion.metrics.IngestionObservabilitySupport; import com.decathlon.idp_core.infrastructure.adapters.ingestion.security.WebhookRequestAuthenticator; +import io.opentelemetry.api.metrics.Meter; import io.opentelemetry.api.trace.Span; import lombok.extern.slf4j.Slf4j; @@ -23,21 +30,33 @@ public class SecurityProcessor { private static final String SPAN_ATTRIBUTE_WEBHOOK_IDENTIFIER = "idp.webhook.identifier"; private static final String SPAN_ATTRIBUTE_SECURITY_RESULT = "idp.webhook.security.result"; private static final String RESULT_SUCCESS = "SUCCESS"; + private static final String RESULT_REJECTED = "REJECTED"; + private static final String RESULT_ERROR = "ERROR"; private final List strategies; + private final Meter meter; - public SecurityProcessor(List strategies) { + public SecurityProcessor(List strategies, + @Qualifier("securityMeter") Meter meter) { this.strategies = List.copyOf(strategies); + this.meter = meter; } public void validate(Map headers, byte[] rawPayload, WebhookConnector webhookConnector) { String webhookIdentifier = webhookConnector.identifier(); - Span.current().setAttribute(SPAN_ATTRIBUTE_WEBHOOK_IDENTIFIER, webhookIdentifier); - WebhookSecurity security = webhookConnector.security(); + String securityType = security == null || security.type() == null + ? WebhookSecurityType.NONE.name() + : security.type().name(); + + incrementCounter(IngestionObservabilityConstants.SECURITY_REQUESTS_TOTAL, ATTRIBUTE_CONNECTOR, + webhookIdentifier, "security_type", securityType); + IngestionObservabilitySupport.setSpanAttribute(Span.current(), + SPAN_ATTRIBUTE_WEBHOOK_IDENTIFIER, webhookIdentifier); + if (security == null || security.type() == WebhookSecurityType.NONE) { - Span.current().setAttribute(SPAN_ATTRIBUTE_SECURITY_RESULT, RESULT_SUCCESS); + recordSecurityResult(webhookIdentifier, securityType, RESULT_SUCCESS); return; } @@ -53,13 +72,37 @@ public void validate(Map headers, byte[] rawPayload, authenticator.validateRequest(headers, rawPayload == null ? new byte[0] : rawPayload, security.config()); - Span.current().setAttribute(SPAN_ATTRIBUTE_SECURITY_RESULT, RESULT_SUCCESS); + recordSecurityResult(webhookIdentifier, securityType, RESULT_SUCCESS); log.debug("Webhook security validation passed for connector '{}' with strategy '{}'.", webhookIdentifier, security.type()); } catch (RuntimeException exception) { - Span.current().setAttribute(SPAN_ATTRIBUTE_SECURITY_RESULT, "REJECTED"); + String result = isSecurityRejected(exception) ? RESULT_REJECTED : RESULT_ERROR; + recordSecurityResult(webhookIdentifier, securityType, result); + incrementCounter(IngestionObservabilityConstants.SECURITY_VALIDATION_ERRORS_TOTAL, + ATTRIBUTE_CONNECTOR, webhookIdentifier, ATTRIBUTE_SECURITY_TYPE, securityType, + ATTRIBUTE_REASON, classifyFailure(exception)); throw exception; } } + private void recordSecurityResult(String webhookIdentifier, String securityType, String result) { + incrementCounter(IngestionObservabilityConstants.SECURITY_RESULT, ATTRIBUTE_CONNECTOR, + webhookIdentifier, ATTRIBUTE_SECURITY_TYPE, securityType, "result", result); + IngestionObservabilitySupport.setSpanAttribute(Span.current(), SPAN_ATTRIBUTE_SECURITY_RESULT, + result); + } + + private void incrementCounter(String metricName, String... attributes) { + IngestionObservabilitySupport.incrementCounter(meter, metricName, attributes); + } + + private boolean isSecurityRejected(RuntimeException exception) { + return exception instanceof WebhookAuthenticationException + || exception instanceof WebhookSecurityException; + } + + private String classifyFailure(RuntimeException exception) { + String reason = exception.getClass().getSimpleName(); + return reason.isBlank() ? "unknown" : reason; + } } diff --git a/src/main/java/com/decathlon/idp_core/infrastructure/adapters/ingestion/processor/decoder/DecodingProcessor.java b/src/main/java/com/decathlon/idp_core/infrastructure/adapters/ingestion/processor/decoder/DecodingProcessor.java index cb010f08..23d73bfe 100644 --- a/src/main/java/com/decathlon/idp_core/infrastructure/adapters/ingestion/processor/decoder/DecodingProcessor.java +++ b/src/main/java/com/decathlon/idp_core/infrastructure/adapters/ingestion/processor/decoder/DecodingProcessor.java @@ -1,6 +1,7 @@ package com.decathlon.idp_core.infrastructure.adapters.ingestion.processor.decoder; import static com.decathlon.idp_core.infrastructure.adapters.ingestion.configuration.IngestionConstants.*; +import static com.decathlon.idp_core.infrastructure.adapters.ingestion.configuration.IngestionObservabilityConstants.*; import java.io.ByteArrayInputStream; import java.io.ByteArrayOutputStream; @@ -14,11 +15,14 @@ import java.util.zip.GZIPInputStream; import java.util.zip.ZipException; +import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Component; import com.decathlon.idp_core.infrastructure.adapters.ingestion.exception.WebhookDecodingException; +import io.opentelemetry.api.metrics.Meter; + /// Processor responsible for decoding webhook payloads based on HTTP `Content-Encoding` headers. /// /// Supported Encodings: @@ -36,20 +40,17 @@ public class DecodingProcessor { private static final int SANITIZED_HEADER_MAX_LENGTH = 128; private static final int DECOMPRESSION_BUFFER_SIZE = 8192; - // Maximum input payload size (before decompression) — aligns with - // PayloadValidationProcessor - // Prevents unbounded String→byte[] or byte[]→String conversions private static final long MAX_INPUT_PAYLOAD_BYTES = 10L * 1024 * 1024; // 10MB private final Map decoders; - - // Max decompressed size — configurable via idp.ingestion.max-decompressed-bytes - // to prevent Zip Bomb attacks. + private final Meter meter; private final long maxDecompressedBytes; public DecodingProcessor( - @Value("${idp.ingestion.max-decompressed-bytes:10485760}") long maxDecompressedBytes) { + @Value("${idp.ingestion.max-decompressed-bytes:10485760}") long maxDecompressedBytes, + @Qualifier("decodingMeter") Meter meter) { this.maxDecompressedBytes = maxDecompressedBytes; + this.meter = meter; this.decoders = Map.of(CONTENT_ENCODING_IDENTITY, this::decodeIdentity, CONTENT_ENCODING_GZIP, this::decodeGzip); } @@ -60,39 +61,86 @@ public DecodingProcessor( public String decode(Object encodedPayload, Map headers) { String contentEncoding = extractContentEncodingHeader(headers); List encodingChain = parseEncodingChain(contentEncoding); + String connector = headers == null + ? "unknown" + : String.valueOf(headers.getOrDefault("connector", "unknown")); - if (encodingChain.isEmpty()) { - return payloadToString(encodedPayload); - } + try { + if (encodingChain.isEmpty()) { + String decoded = payloadToString(encodedPayload); + recordPayloadSizeMetric(connector, decoded.getBytes(StandardCharsets.UTF_8).length); + incrementDecodedMetric(connector, "identity"); + return decoded; + } - // Reject unsupported encodings explicitly — silent pass-through risks silent - // data corruption. - List unsupported = encodingChain.stream() - .filter(encoding -> !decoders.containsKey(encoding)).toList(); - if (!unsupported.isEmpty()) { - throw new WebhookDecodingException( - "Unsupported Content-Encoding: '" + sanitizeHeaderValue(contentEncoding) - + "'. Supported encodings: " + String.join(", ", decoders.keySet())); - } + // Reject unsupported encodings explicitly — silent pass-through risks silent + // data corruption. + List unsupported = encodingChain.stream() + .filter(encoding -> !decoders.containsKey(encoding)).toList(); + if (!unsupported.isEmpty()) { + String reason = "unsupported_encoding"; + incrementDecodeErrorMetric(connector, reason); + throw new WebhookDecodingException( + "Unsupported Content-Encoding: '" + sanitizeHeaderValue(contentEncoding) + + "'. Supported encodings: " + String.join(", ", decoders.keySet())); + } + + byte[] decodedPayload = toByteArray(encodedPayload); + if (decodedPayload.length == 0 && encodingChain.contains(CONTENT_ENCODING_GZIP)) { + incrementDecodeErrorMetric(connector, "empty_gzip_payload"); + throw new WebhookDecodingException("Empty payload cannot be decoded as gzip"); + } - byte[] decodedPayload = toByteArray(encodedPayload); - if (decodedPayload.length == 0 && encodingChain.contains(CONTENT_ENCODING_GZIP)) { - throw new WebhookDecodingException("Empty payload cannot be decoded as gzip"); + return decodePayloadWithEncodingChain(decodedPayload, encodingChain, contentEncoding, + connector); + } catch (RuntimeException exception) { + if (!(exception instanceof WebhookDecodingException)) { + incrementDecodeErrorMetric(connector, exception.getClass().getSimpleName()); + } + throw exception; } + } + private String decodePayloadWithEncodingChain(byte[] decodedPayload, List encodingChain, + String contentEncoding, String connector) { try { for (int i = encodingChain.size() - 1; i >= 0; i--) { decodedPayload = decoders.get(encodingChain.get(i)).decode(decodedPayload); } - return new String(decodedPayload, StandardCharsets.UTF_8); + String result = new String(decodedPayload, StandardCharsets.UTF_8); + recordPayloadSizeMetric(connector, decodedPayload.length); + incrementDecodedMetric(connector, String.join(",", encodingChain)); + return result; } catch (ZipException e) { + incrementDecodeErrorMetric(connector, "corrupted_gzip_payload"); throw new WebhookDecodingException("Corrupted or invalid compressed gzip stream", e); } catch (IOException e) { + incrementDecodeErrorMetric(connector, "decompression_failure"); throw new WebhookDecodingException( "Failed to decompress payload for encoding: " + sanitizeHeaderValue(contentEncoding), e); } } + private void recordPayloadSizeMetric(String connector, long payloadSizeBytes) { + meter.histogramBuilder(PAYLOAD_SIZE_BYTES).ofLongs().build().record(payloadSizeBytes, + io.opentelemetry.api.common.Attributes + .of(io.opentelemetry.api.common.AttributeKey.stringKey("connector"), connector)); + } + + private void incrementDecodedMetric(String connector, String encoding) { + meter.counterBuilder(PAYLOAD_DECODED_TOTAL).build().add(1, + io.opentelemetry.api.common.Attributes.of( + io.opentelemetry.api.common.AttributeKey.stringKey(ATTRIBUTE_CONNECTOR), connector, + io.opentelemetry.api.common.AttributeKey.stringKey("encoding"), encoding)); + } + + private void incrementDecodeErrorMetric(String connector, String reason) { + meter.counterBuilder(PAYLOAD_DECODE_ERRORS_TOTAL).build().add(1, + io.opentelemetry.api.common.Attributes.of( + io.opentelemetry.api.common.AttributeKey.stringKey(ATTRIBUTE_CONNECTOR), connector, + io.opentelemetry.api.common.AttributeKey.stringKey("reason"), reason)); + } + private String extractContentEncodingHeader(Map headers) { if (headers == null) return null; diff --git a/src/main/java/com/decathlon/idp_core/infrastructure/adapters/ingestion/route/GenericInboundEventRouteBuilder.java b/src/main/java/com/decathlon/idp_core/infrastructure/adapters/ingestion/route/GenericInboundEventRouteBuilder.java index df730761..3472ddf6 100644 --- a/src/main/java/com/decathlon/idp_core/infrastructure/adapters/ingestion/route/GenericInboundEventRouteBuilder.java +++ b/src/main/java/com/decathlon/idp_core/infrastructure/adapters/ingestion/route/GenericInboundEventRouteBuilder.java @@ -2,9 +2,12 @@ import static com.decathlon.idp_core.infrastructure.adapters.ingestion.configuration.IngestionConstants.*; +import java.util.stream.Collectors; + import org.apache.camel.Exchange; import org.apache.camel.LoggingLevel; import org.apache.camel.builder.RouteBuilder; +import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Component; import com.decathlon.idp_core.domain.exception.webhook.WebhookConfigurationMissingException; @@ -17,11 +20,12 @@ import com.decathlon.idp_core.infrastructure.adapters.ingestion.processor.SecurityProcessor; import com.decathlon.idp_core.infrastructure.adapters.ingestion.processor.decoder.DecodingProcessor; -import lombok.RequiredArgsConstructor; +import io.opentelemetry.api.trace.Span; +import lombok.extern.slf4j.Slf4j; /// Generic Camel Route pipeline handling inbound webhook fetching, status validation, and payload decoding. @Component -@RequiredArgsConstructor +@Slf4j public class GenericInboundEventRouteBuilder extends RouteBuilder { private final WebhookConnectorService webhookConnectorService; @@ -31,15 +35,35 @@ public class GenericInboundEventRouteBuilder extends RouteBuilder { private final PayloadValidationProcessor payloadValidationProcessor; private final WebhookExceptionRouteBuilder webhookExceptionRouteBuilder; + private final boolean tracingDiagnosticEnabled; + + public GenericInboundEventRouteBuilder(WebhookConnectorService webhookConnectorService, + DecodingProcessor decodingProcessor, SecurityProcessor securityProcessor, + IngestionProcessor ingestionProcessor, PayloadValidationProcessor payloadValidationProcessor, + WebhookExceptionRouteBuilder webhookExceptionRouteBuilder, + @Value("${app.ingestion.tracing-diagnostic-enabled:false}") boolean tracingDiagnosticEnabled) { + this.webhookConnectorService = webhookConnectorService; + this.decodingProcessor = decodingProcessor; + this.securityProcessor = securityProcessor; + this.ingestionProcessor = ingestionProcessor; + this.payloadValidationProcessor = payloadValidationProcessor; + this.webhookExceptionRouteBuilder = webhookExceptionRouteBuilder; + this.tracingDiagnosticEnabled = tracingDiagnosticEnabled; + log.info("GenericInboundEventRouteBuilder initialized with tracingDiagnosticEnabled={}", + tracingDiagnosticEnabled); + } + @Override public void configure() throws Exception { webhookExceptionRouteBuilder.configureExceptions(this); // Main Pipeline - from(DIRECT_PROCESS_EVENT).routeId(ROUTE_ID_WEBHOOK_PIPELINE).to(DIRECT_FETCH_CONFIGURATION) - .to(DIRECT_VALIDATE_ENABLED).process(payloadValidationProcessor::validate) - .to(DIRECT_VALIDATE_SECURITY).to(DIRECT_DECODE_PAYLOAD).to(DIRECT_INGEST_PAYLOAD) - .removeHeaders("*").setHeader(Exchange.HTTP_RESPONSE_CODE, constant(HTTP_CREATED)) + from(DIRECT_PROCESS_EVENT).routeId(ROUTE_ID_WEBHOOK_PIPELINE) + .process(exchange -> logTracingDiagnostics(exchange, "direct-process-event-entry")) + .to(DIRECT_FETCH_CONFIGURATION).to(DIRECT_VALIDATE_ENABLED) + .process(payloadValidationProcessor::validate).to(DIRECT_VALIDATE_SECURITY) + .to(DIRECT_DECODE_PAYLOAD).to(DIRECT_INGEST_PAYLOAD).removeHeaders("*") + .setHeader(Exchange.HTTP_RESPONSE_CODE, constant(HTTP_CREATED)) .setHeader(Exchange.CONTENT_TYPE, constant(APPLICATION_JSON)) .setBody(constant(SUCCESS_WEBHOOK_EVENT_PROCESSED)); @@ -107,6 +131,7 @@ public void configure() throws Exception { from(DIRECT_INGEST_PAYLOAD).routeId(ROUTE_ID_INGEST_PAYLOAD) .log(LoggingLevel.DEBUG, "Ingesting payload for webhook ID: ${exchangeProperty.connectorIdentifier}") + .process(exchange -> logTracingDiagnostics(exchange, "direct-ingest-payload")) .process(exchange -> { String decodedPayload = exchange.getIn().getBody(String.class); WebhookConnector config = exchange.getProperty(WEBHOOK_CONFIG_PROPERTY, @@ -114,4 +139,34 @@ public void configure() throws Exception { ingestionProcessor.ingest(decodedPayload, config); }); } + + private void logTracingDiagnostics(Exchange exchange, String stage) { + if (!tracingDiagnosticEnabled || !log.isInfoEnabled()) { + return; + } + + Span currentSpan = Span.current(); + String traceId = currentSpan.getSpanContext().getTraceId(); + String spanId = currentSpan.getSpanContext().getSpanId(); + boolean validSpan = currentSpan.getSpanContext().isValid(); + + String interestingProperties = exchange.getProperties().entrySet().stream() + .filter(entry -> isTelemetryKey(entry.getKey())) + .map(entry -> entry.getKey() + "=" + + (entry.getValue() == null ? "null" : entry.getValue().getClass().getSimpleName())) + .sorted().collect(Collectors.joining(", ")); + + log.info( + "[OTEL CAMEL DIAG] stage={} routeId={} exchangeId={} spanValid={} traceId={} spanId={} traceparent={} b3={} telemetryProperties=[{}]", + stage, exchange.getFromRouteId(), exchange.getExchangeId(), validSpan, traceId, spanId, + exchange.getIn().getHeader("traceparent"), exchange.getIn().getHeader("b3"), + interestingProperties); + } + + private boolean isTelemetryKey(String key) { + String normalized = key == null ? "" : key.toLowerCase(); + return normalized.contains("telemetry") || normalized.contains("trace") + || normalized.contains("span") || normalized.contains("baggage") + || normalized.startsWith("camel"); + } } diff --git a/src/main/resources/application.yml b/src/main/resources/application.yml index 83ac46c2..f02b7a34 100644 --- a/src/main/resources/application.yml +++ b/src/main/resources/application.yml @@ -1,4 +1,3 @@ ---- spring: web: resources: @@ -6,7 +5,6 @@ spring: add-mappings: false cors: # Exact origins — set via SPRING_WEB_CORS_ALLOWED_ORIGINS env var (comma-separated). - # Use for known, fixed origins. allowed-origins: ${SPRING_WEB_CORS_ALLOWED_ORIGINS:} # Pattern-based origins — set via SPRING_WEB_CORS_ALLOWED_ORIGIN_PATTERNS env var (comma-separated). allowed-origin-patterns: ${SPRING_WEB_CORS_ALLOWED_ORIGIN_PATTERNS:} @@ -112,6 +110,8 @@ springdoc: app: ingestion: + # Enable temporary diagnostics to inspect Camel exchange telemetry context propagation. + tracing-diagnostic-enabled: ${APP_INGESTION_TRACING_DIAGNOSTIC_ENABLED:true} webhook: base-path: /webhooks identifier-path: /{webhookIdentifier} @@ -169,6 +169,9 @@ idp: allowed-jwks-hosts: ${ALLOWED_JWKS_HOSTS:} logging: + pattern: + # Keep trace/span ids in logs with the OpenTelemetry starter. + correlation: "[%32X{traceId:-}-%16X{spanId:-}] " level: # Application log level, defaults to INFO. Override per environment via LOG_LEVEL env var. com.decathlon: ${LOG_LEVEL:INFO} @@ -176,50 +179,78 @@ logging: org.springframework.context.support: WARN # Suppresses a specific repetitive bean registration delegate log line. org.springframework.context.support.PostProcessorRegistrationDelegate: ERROR + io.opentelemetry.sdk: DEBUG + io.opentelemetry.exporter.otlp: DEBUG + com.decathlon.idp_core.infrastructure.adapters.ingestion: DEBUG + io.opentelemetry: TRACE + io.opentelemetry.sdk.metrics: TRACE + # Affiche la création et l'envoi des Spans OTel + io.opentelemetry.sdk.trace: TRACE + # Affiche l'interception des requêtes par Camel OpenTelemetry + org.apache.camel.opentelemetry: TRACE + io.opentelemetry.exporter: TRACE + io.opentelemetry.exporter.sender.okhttp: TRACE + khttp3: TRACE - -management: - tracing: - sampling: - # Fraction of traces kept (1.0 = 100%). Standard OTel variable. - probability: ${OTEL_TRACES_SAMPLER_ARG:0.1} - export: - enabled: ${OTEL_TRACING_ENABLED:false} - - # Trace OTLP export lives under management.opentelemetry.tracing.export.otlp.* (see below). - otlp: - metrics: - export: - # Metrics export is opt-in. Signal-specific variables win over the generic ones. - url: ${OTEL_EXPORTER_OTLP_METRICS_ENDPOINT:${OTEL_EXPORTER_OTLP_ENDPOINT:http://localhost:4318}/v1/metrics} - enabled: ${OTEL_METRICS_EXPORT_ENABLED:false} - transport: ${OTEL_EXPORTER_OTLP_METRICS_PROTOCOL:${OTEL_EXPORTER_OTLP_PROTOCOL:http/protobuf}} - opentelemetry: - tracing: - export: - otlp: - # Boot does not append /v1/traces itself: the full traces URL is required for HTTP. - endpoint: ${OTEL_EXPORTER_OTLP_TRACES_ENDPOINT:${OTEL_EXPORTER_OTLP_ENDPOINT:http://localhost:4318}/v1/traces} - # Boot's transport enum only accepts http or grpc (not the OTel value http/protobuf). - transport: ${OTEL_EXPORTER_OTLP_TRACES_TRANSPORT:${OTEL_EXPORTER_OTLP_PROTOCOL:http}} - # Extra attributes (for example X-Datadog-Org, X-Backend) are set by the user - # through OTEL_RESOURCE_ATTRIBUTES, not hard-coded here. - resource-attributes: +# OpenTelemetry Spring Boot Starter Configuration +otel: + register-global: true + service: + name: ${OTEL_SERVICE_NAME:internal-developer-platform} + resource: + attributes: deployment.environment: ${OTEL_ENV:local} service.namespace: ${OTEL_SERVICE_NAMESPACE:idp-core} - service.name: ${OTEL_SERVICE_NAME:internal-developer-platform} + metrics: + exporter: ${OTEL_METRICS_EXPORTER:otlp} + + metric: + export: + interval: ${OTEL_METRIC_EXPORT_INTERVAL:60000} + + traces: + exporter: ${OTEL_TRACES_EXPORTER:otlp} + sampler: ${OTEL_TRACES_SAMPLER:always_on} + sampler-arg: ${OTEL_TRACES_SAMPLER_ARG:1.0} + + logs: + exporter: ${OTEL_LOGS_EXPORTER:none} + + exporter: + otlp: + endpoint: ${OTEL_EXPORTER_OTLP_ENDPOINT:http://localhost:4318} + protocol: ${OTEL_EXPORTER_OTLP_PROTOCOL:http/protobuf} + metrics: + endpoint: ${OTEL_EXPORTER_OTLP_METRICS_ENDPOINT:${OTEL_EXPORTER_OTLP_ENDPOINT:http://localhost:4318}/v1/metrics} + protocol: ${OTEL_EXPORTER_OTLP_METRICS_PROTOCOL:${OTEL_EXPORTER_OTLP_PROTOCOL:http/protobuf}} + traces: + endpoint: ${OTEL_EXPORTER_OTLP_TRACES_ENDPOINT:${OTEL_EXPORTER_OTLP_ENDPOINT:http://localhost:4318}/v1/traces} + protocol: ${OTEL_EXPORTER_OTLP_TRACES_PROTOCOL:${OTEL_EXPORTER_OTLP_PROTOCOL:http/protobuf}} + instrumentation: + common: + logging: + # Spring log patterns use traceId/spanId while OTel defaults to trace_id/span_id. + trace-id-key: traceId + span-id-key: spanId + micrometer: + enabled: false + spring-web: + enabled: true + spring-webmvc: + enabled: true + # OTEL_INSTRUMENTATION_JDBC_ENABLED=true to add database spans. + jdbc: + enabled: ${OTEL_INSTRUMENTATION_JDBC_ENABLED:false} + +management: endpoints: web: exposure: - # Explicitly restrict exposed actuator endpoints to health only. - # Prevents accidental exposure of env, beans, configprops, etc. include: health endpoint: health: probes: - # Enables liveness (/actuator/health/liveness) and - # readiness (/actuator/health/readiness) probes. enabled: true group: liveness: @@ -238,7 +269,10 @@ management: # Avoids leaking internal infrastructure info to unauthenticated callers. show-details: when_authorized - camel: opentelemetry2: enabled: true + # Keep core route spans enabled; processor spans depend on them for active scope propagation. + disable-core-processors: false + # Keep processor-level spans (setHeader, setBody, processors...) for ingestion diagnostics. + trace-processors: true diff --git a/src/test/java/com/decathlon/idp_core/infrastructure/adapters/ingestion/processor/DecodingProcessorTest.java b/src/test/java/com/decathlon/idp_core/infrastructure/adapters/ingestion/processor/DecodingProcessorTest.java index f5ec5ebc..56f5f206 100644 --- a/src/test/java/com/decathlon/idp_core/infrastructure/adapters/ingestion/processor/DecodingProcessorTest.java +++ b/src/test/java/com/decathlon/idp_core/infrastructure/adapters/ingestion/processor/DecodingProcessorTest.java @@ -20,6 +20,10 @@ import com.decathlon.idp_core.infrastructure.adapters.ingestion.exception.WebhookDecodingException; import com.decathlon.idp_core.infrastructure.adapters.ingestion.processor.decoder.DecodingProcessor; +import io.opentelemetry.api.metrics.Meter; +import io.opentelemetry.sdk.OpenTelemetrySdk; +import io.opentelemetry.sdk.metrics.SdkMeterProvider; + @DisplayName("DecodingProcessor unit tests") class DecodingProcessorTest { @@ -27,9 +31,13 @@ class DecodingProcessorTest { private static final byte[] CORRUPTED_GZIP_PAYLOAD = "not-a-gzip-stream" .getBytes(StandardCharsets.UTF_8); + private final Meter meter = OpenTelemetrySdk.builder() + .setMeterProvider(SdkMeterProvider.builder().build()).build() + .getMeter("com.decathlon.idp_core.tests"); + // Default processor with the production default limit (10 MB) private final DecodingProcessor decodingProcessor = new DecodingProcessor( - IngestionConstants.MAX_DECOMPRESSED_BYTES); + IngestionConstants.MAX_DECOMPRESSED_BYTES, meter); @ParameterizedTest(name = "[{index}]") @MethodSource("passThroughCases") @@ -103,7 +111,7 @@ void decode_throwsWebhookDecodingException_whenContentEncodingIsUnsupported( @DisplayName("Throws WebhookDecodingException when decompressed payload exceeds the configured size limit (Zip Bomb protection)") void decode_throwsWebhookDecodingException_whenDecompressedSizeExceedsLimit() throws Exception { // Processor with a very small limit of 10 bytes to trigger the zip bomb guard - DecodingProcessor restrictedProcessor = new DecodingProcessor(10L); + DecodingProcessor restrictedProcessor = new DecodingProcessor(10L, meter); byte[] gzipPayload = gzip("This payload will exceed the 10 bytes decompression limit"); WebhookDecodingException exception = assertThrows(WebhookDecodingException.class, diff --git a/src/test/java/com/decathlon/idp_core/infrastructure/adapters/ingestion/processor/IngestionProcessorTest.java b/src/test/java/com/decathlon/idp_core/infrastructure/adapters/ingestion/processor/IngestionProcessorTest.java index 67855a54..ea56bd10 100644 --- a/src/test/java/com/decathlon/idp_core/infrastructure/adapters/ingestion/processor/IngestionProcessorTest.java +++ b/src/test/java/com/decathlon/idp_core/infrastructure/adapters/ingestion/processor/IngestionProcessorTest.java @@ -41,6 +41,11 @@ import ch.qos.logback.classic.Logger; import ch.qos.logback.classic.spi.ILoggingEvent; import ch.qos.logback.core.read.ListAppender; +import io.opentelemetry.api.OpenTelemetry; +import io.opentelemetry.api.metrics.Meter; +import io.opentelemetry.sdk.OpenTelemetrySdk; +import io.opentelemetry.sdk.metrics.SdkMeterProvider; +import io.opentelemetry.sdk.testing.exporter.InMemoryMetricReader; /// Unit tests for IngestionProcessor covering all mapping actions and edge cases. /// @@ -70,7 +75,14 @@ void setUp() { listAppender = new ListAppender<>(); listAppender.start(); logger.addAppender(listAppender); - ingestionProcessor = new IngestionProcessor(mappingEngine, entityService); + + InMemoryMetricReader metricReader = InMemoryMetricReader.create(); + OpenTelemetry openTelemetry = OpenTelemetrySdk.builder() + .setMeterProvider(SdkMeterProvider.builder().registerMetricReader(metricReader).build()) + .build(); + Meter meter = openTelemetry.getMeter("com.decathlon.idp_core.tests"); + + ingestionProcessor = new IngestionProcessor(mappingEngine, entityService, meter); } private Entity createTestEntity(String templateId, String identifier) { diff --git a/src/test/resources/application-test.yml b/src/test/resources/application-test.yml index 22564216..301761f9 100644 --- a/src/test/resources/application-test.yml +++ b/src/test/resources/application-test.yml @@ -60,10 +60,15 @@ app: hmac-env-value: test-github-secret jwks-env-key: JWKS_URI jwks-env-value: https://auth.example.com/.well-known/jwks.json +otel: + traces: + exporter: none + metrics: + exporter: none + logs: + exporter: none + management: - # Tests must not create or export traces. - tracing: - enabled: false endpoint: health: probes: