附录：随书实验完整代码

name: observability-book
services:
  redis:
    image: library/redis:7.2.5-alpine@sha256:6aaf3f5e6bc8a592fbfe2cccf19eb36d27c39d12dab4f4b01556b7449e7b1f44
    ports: ["127.0.0.1:6379:6379"]
  namesrv:
    image: apache/rocketmq:5.3.2@sha256:4a80ff5cd4b0a3d3ef6dc338ce2028a85bd995648cb3aa5c3e753746d2dbc587
    command: ["sh", "mqnamesrv"]
    environment:
      JAVA_OPT_EXT: "-Xms128m -Xmx128m -Xmn64m"
    ports: ["127.0.0.1:9876:9876"]
  broker:
    image: apache/rocketmq:5.3.2@sha256:4a80ff5cd4b0a3d3ef6dc338ce2028a85bd995648cb3aa5c3e753746d2dbc587
    command: ["sh", "mqbroker", "-n", "namesrv:9876", "-c", "/home/rocketmq/rocketmq-5.3.2/conf/broker.conf"]
    environment:
      JAVA_OPT_EXT: "-Xms256m -Xmx256m -Xmn128m"
    volumes: ["./config/broker.conf:/home/rocketmq/rocketmq-5.3.2/conf/broker.conf:ro"]
    ports: ["127.0.0.1:10911:10911", "127.0.0.1:10909:10909"]
    depends_on: [namesrv]
  tempo:
    image: grafana/tempo:2.6.1@sha256:ef4384fce6e8ad22b95b243d8fc165628cda655376fd50e7850536ad89d71d50
    command: ["-config.file=/etc/tempo.yaml"]
    volumes: ["./config/tempo.yaml:/etc/tempo.yaml:ro"]
    ports: ["127.0.0.1:3200:3200"]
  collector:
    image: otel/opentelemetry-collector-contrib:0.111.0@sha256:a2a52e43c1a80aa94120ad78c2db68780eb90e6d11c8db5b3ce2f6a0cc6b5029
    command: ["--config=/etc/collector.yaml"]
    volumes: ["./config/collector.yaml:/etc/collector.yaml:ro"]
    ports: ["127.0.0.1:4318:4318", "127.0.0.1:8889:8889", "127.0.0.1:13133:13133"]
    depends_on: [tempo]
  prometheus:
    image: prom/prometheus:v2.55.1@sha256:2659f4c2ebb718e7695cb9b25ffa7d6be64db013daba13e05c875451cf51b0d3
    command: ["--config.file=/etc/prometheus/prometheus.yaml", "--storage.tsdb.path=/prometheus", "--enable-feature=exemplar-storage"]
    volumes: ["./config/prometheus.yaml:/etc/prometheus/prometheus.yaml:ro", "./config/alerts.yaml:/etc/prometheus/alerts.yaml:ro"]
    ports: ["127.0.0.1:9095:9090"]
    depends_on: [collector]

groups:
  - name: book-slo
    rules:
      - alert: BookFastBurn
        expr: (sum(rate(lab_requests_total{service_name="book-gateway",outcome="error"}[1m])) / sum(rate(lab_requests_total{service_name="book-gateway"}[1m])) > 0.05) and (sum(increase(lab_requests_total{service_name="book-gateway"}[1m])) >= 5)
        for: 2s
        labels:
          severity: exercise
        annotations:
          summary: "实验请求错误率超过5%，为99%可用性目标的5倍预算消耗率"

receivers:
  otlp:
    protocols:
      http:
        endpoint: 0.0.0.0:4318
processors:
  memory_limiter:
    check_interval: 1s
    limit_mib: 200
    spike_limit_mib: 50
  batch:
    timeout: 1s
    send_batch_size: 128
exporters:
  otlp/tempo:
    endpoint: tempo:4317
    tls:
      insecure: true
  prometheus:
    enable_open_metrics: true
    endpoint: 0.0.0.0:8889
    resource_to_telemetry_conversion:
      enabled: true
  debug:
    verbosity: detailed
extensions:
  health_check:
    endpoint: 0.0.0.0:13133
service:
  extensions: [health_check]
  pipelines:
    traces:
      receivers: [otlp]
      processors: [memory_limiter, batch]
      exporters: [otlp/tempo]
    metrics:
      receivers: [otlp]
      processors: [memory_limiter, batch]
      exporters: [prometheus]
    logs:
      receivers: [otlp]
      processors: [memory_limiter, batch]
      exporters: [debug]

global:
  scrape_interval: 1s
  evaluation_interval: 1s
rule_files: ["/etc/prometheus/alerts.yaml"]
scrape_configs:
  - job_name: book-collector
    static_configs:
      - targets: ["collector:8889"]

server:
  http_listen_port: 3200
distributor:
  receivers:
    otlp:
      protocols:
        grpc:
          endpoint: 0.0.0.0:4317
        http:
          endpoint: 0.0.0.0:4318
ingester:
  trace_idle_period: 2s
  max_block_duration: 1m
compactor:
  compaction:
    block_retention: 1h
storage:
  trace:
    backend: local
    wal:
      path: /tmp/tempo/wal
    local:
      path: /tmp/tempo/blocks

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion><groupId>book.observability</groupId><artifactId>observability-lab</artifactId><version>1.0.0</version>
  <properties><maven.compiler.release>21</maven.compiler.release><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding><otel.version>1.55.0</otel.version></properties>
  <dependencies>
    <dependency><groupId>io.opentelemetry</groupId><artifactId>opentelemetry-sdk</artifactId><version>${otel.version}</version></dependency>
    <dependency><groupId>io.opentelemetry</groupId><artifactId>opentelemetry-exporter-otlp</artifactId><version>${otel.version}</version></dependency>
    <dependency><groupId>redis.clients</groupId><artifactId>jedis</artifactId><version>5.2.0</version></dependency>
    <dependency><groupId>org.apache.rocketmq</groupId><artifactId>rocketmq-client</artifactId><version>5.3.2</version></dependency>
    <dependency><groupId>com.fasterxml.jackson.core</groupId><artifactId>jackson-databind</artifactId><version>2.17.2</version></dependency>
    <dependency><groupId>org.slf4j</groupId><artifactId>slf4j-simple</artifactId><version>2.0.16</version></dependency>
  </dependencies>
  <build><finalName>observability-lab</finalName><plugins>
    <plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-compiler-plugin</artifactId><version>3.13.0</version></plugin>
    <plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-shade-plugin</artifactId><version>3.6.0</version><executions><execution><phase>package</phase><goals><goal>shade</goal></goals><configuration><createDependencyReducedPom>false</createDependencyReducedPom><filters><filter><artifact>*:*</artifact><excludes><exclude>META-INF/*.SF</exclude><exclude>META-INF/*.DSA</exclude><exclude>META-INF/*.RSA</exclude></excludes></filter></filters><transformers><transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/><transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer"><mainClass>book.SystemLab</mainClass></transformer></transformers></configuration></execution></executions></plugin>
  </plugins></build>
</project>

#!/usr/bin/env bash
set -euo pipefail
cd "$(dirname "$0")/.."
mkdir -p evidence
compose=(docker compose -f lab/compose.yaml)
app_pid=
cleanup() {
  if [[ -n "$app_pid" ]]; then kill "$app_pid" 2>/dev/null || true; wait "$app_pid" 2>/dev/null || true; fi
  "${compose[@]}" logs --no-color > evidence/containers.log 2>&1 || true
  "${compose[@]}" ps -a > evidence/containers-status.txt || true
  "${compose[@]}" down -v --remove-orphans || true
}
trap cleanup EXIT
git rev-parse HEAD > evidence/commit.txt
java -version 2> evidence/java-version.txt
"${compose[@]}" config --images > evidence/images.txt
"${compose[@]}" up -d
topic_ready=0
for ((attempt=0;attempt<90;attempt++)); do
  "${compose[@]}" exec -T broker sh -c 'sh /home/rocketmq/rocketmq-5.3.2/bin/mqadmin updateTopic -n namesrv:9876 -b 127.0.0.1:10911 -t otel-book-orders' > evidence/topic.txt 2>&1 || true
  if grep -qi 'create topic.*success' evidence/topic.txt; then
    "${compose[@]}" exec -T broker sh -c 'sh /home/rocketmq/rocketmq-5.3.2/bin/mqadmin topicRoute -n namesrv:9876 -t otel-book-orders' > evidence/topic-route.json 2>&1 || true
    if grep -q 'brokerAddrs' evidence/topic-route.json; then topic_ready=1; break; fi
  fi
  sleep 2
done
[[ "$topic_ready" == 1 ]] || { cat evidence/topic.txt; exit 1; }
"${compose[@]}" exec -T redis redis-cli PING > evidence/redis-ping.txt
java -jar lab/target/observability-lab.jar > evidence/application.log 2>&1 &
app_pid=$!
python3 tests/verify_system.py --phase full
"${compose[@]}" logs --no-color collector > evidence/collector.log
python3 - <<'PY'
import json
from pathlib import Path
results=json.loads(Path('evidence/results.json').read_text())
trace=results['checks']['normal']['response']['trace_id']
logs=Path('evidence/collector.log').read_text().lower()
assert trace.lower() in logs, 'Collector did not receive trace-correlated OTLP logs'
print('PASS actual OTLP logs contain the same trace id as Tempo spans')
PY
"${compose[@]}" stop collector
python3 tests/verify_system.py --phase collector-outage
"${compose[@]}" start collector
python3 tests/verify_system.py --phase recovered
kill "$app_pid"
wait "$app_pid" || true
app_pid=
SAMPLER_RATIO=0 java -jar lab/target/observability-lab.jar > evidence/application-sampling-zero.log 2>&1 &
app_pid=$!
python3 tests/verify_system.py --phase sampled-out
"${compose[@]}" exec -T redis redis-cli --scan --pattern 'otelbook:*' > evidence/redis-keys.txt

package book;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.sun.net.httpserver.HttpExchange;
import com.sun.net.httpserver.HttpServer;
import io.opentelemetry.api.common.Attributes;
import io.opentelemetry.api.logs.Severity;
import io.opentelemetry.api.trace.Span;
import io.opentelemetry.api.trace.SpanKind;
import io.opentelemetry.api.trace.StatusCode;
import io.opentelemetry.context.Context;
import io.opentelemetry.context.Scope;
import io.opentelemetry.context.propagation.TextMapGetter;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import redis.clients.jedis.JedisPooled;
import java.net.InetSocketAddress;
import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.nio.charset.StandardCharsets;
import java.time.Instant;
import java.time.Duration;
import java.util.Map;
import java.util.List;
import java.util.LinkedHashMap;
import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.Executors;

/** A deliberately small real HTTP -> Redis -> RocketMQ -> Redis system. */
public final class SystemLab {
    static final ObjectMapper JSON = new ObjectMapper();
    static final Telemetry GATEWAY = new Telemetry("book-gateway");
    static final Telemetry ORDERS = new Telemetry("book-orders");
    static final Telemetry WORKER = new Telemetry("book-worker");
    static final JedisPooled REDIS = new JedisPooled(System.getenv().getOrDefault("REDIS_HOST", "127.0.0.1"), 6379);
    static final DefaultMQProducer PRODUCER = new DefaultMQProducer("otel-book-producer");
    static final DefaultMQPushConsumer CONSUMER = new DefaultMQPushConsumer("otel-book-worker");
    static final HttpClient HTTP = HttpClient.newBuilder().connectTimeout(Duration.ofSeconds(3)).build();
    static final ConcurrentLinkedDeque<Map<String,Object>> EVENTS = new ConcurrentLinkedDeque<>();
    static final String TOPIC = "otel-book-orders";
    static final TextMapGetter<HttpExchange> HTTP_GETTER = new TextMapGetter<>() {
        public Iterable<String> keys(HttpExchange exchange) { return exchange.getRequestHeaders().keySet(); }
        public String get(HttpExchange exchange, String key) { return exchange.getRequestHeaders().getFirst(key); }
    };
    static final TextMapGetter<MessageExt> MQ_GETTER = new TextMapGetter<>() {
        public Iterable<String> keys(MessageExt message) { return message.getProperties().keySet(); }
        public String get(MessageExt message, String key) { return message.getProperty(key); }
    };
    static byte[] bytes(Object value) throws Exception { return JSON.writeValueAsBytes(value); }
    static void respond(HttpExchange exchange, int status, Object body) throws Exception {
        byte[] raw = bytes(body);
        exchange.getResponseHeaders().set("Content-Type", "application/json; charset=utf-8");
        exchange.getResponseHeaders().set("X-Trace-Id", Span.current().getSpanContext().getTraceId());
        exchange.sendResponseHeaders(status, raw.length);
        exchange.getResponseBody().write(raw);
    }
    static void event(Telemetry telemetry, String name, String id) {
        Map<String,Object> row = new LinkedHashMap<>();
        row.put("timestamp", Instant.now().toString());row.put("service",telemetry.service);row.put("event",name);
        row.put("order_id",id);row.put("trace_id",Span.current().getSpanContext().getTraceId());
        row.put("span_id",Span.current().getSpanContext().getSpanId());row.put("sampled",Span.current().getSpanContext().isSampled());
        EVENTS.add(row);while(EVENTS.size()>300)EVENTS.pollFirst();
        try { System.out.println(JSON.writeValueAsString(row)); } catch(Exception error) { throw new RuntimeException(error); }
        telemetry.logger.logRecordBuilder().setContext(Context.current()).setSeverity(Severity.INFO)
                .setBody(name).setAttribute("order.id",id).emit();
    }
    static int delay(JsonNode body, String name) {
        int value=body.path(name).asInt(0);
        if(value<0||value>1500)throw new IllegalArgumentException("Delay outside 0..1500 ms");
        return value;
    }
    static void consume(MessageExt message) throws Exception {
        Context parent = WORKER.sdk.getPropagators().getTextMapPropagator().extract(Context.root(),message,MQ_GETTER);
        Span span = WORKER.tracer.spanBuilder("rocketmq process " + TOPIC).setSpanKind(SpanKind.CONSUMER)
                .setParent(parent).setAttribute("messaging.system","rocketmq").setAttribute("messaging.destination.name",TOPIC).startSpan();
        long start=System.nanoTime();
        try(Scope scope=span.makeCurrent()) {
            JsonNode body=JSON.readTree(message.getBody());String id=body.path("id").asText();
            span.setAttribute("order.id",id);Thread.sleep(body.path("worker_delay_ms").asInt());
            Span redis=WORKER.tracer.spanBuilder("redis SET processed").setSpanKind(SpanKind.CLIENT).startSpan();
            try(Scope ignored=redis.makeCurrent()) {REDIS.setex("otelbook:processed:"+id,3600,JSON.writeValueAsString(Map.of("id",id,"trace_id",span.getSpanContext().getTraceId())));}
            finally {redis.end();}
            event(WORKER,"worker.processed",id);
            WORKER.requests.add(1,Attributes.builder().put("route","message").put("outcome","ok").build());
        } catch(Exception error) {span.recordException(error);span.setStatus(StatusCode.ERROR);throw error;}
        finally {WORKER.duration.record((System.nanoTime()-start)/1e9,Attributes.of(io.opentelemetry.api.common.AttributeKey.stringKey("route"),"message"),Context.root().with(span));span.end();}
    }
    static void order(HttpExchange exchange) {
        Context parent=ORDERS.sdk.getPropagators().getTextMapPropagator().extract(Context.root(),exchange,HTTP_GETTER);
        Span span=ORDERS.tracer.spanBuilder("orders " + exchange.getRequestMethod() + " /orders").setParent(parent)
                .setSpanKind(SpanKind.SERVER).setAttribute("http.route","/orders").startSpan();
        long start=System.nanoTime();String outcome="ok";
        try(Scope scope=span.makeCurrent()) {
            if(exchange.getRequestMethod().equals("GET")) {
                String id=exchange.getRequestURI().getPath().replaceFirst("^/orders/","");
                Span redis=ORDERS.tracer.spanBuilder("redis GET order").setSpanKind(SpanKind.CLIENT).startSpan();
                String value;try(Scope ignored=redis.makeCurrent()){value=REDIS.get("otelbook:order:"+id);}finally{redis.end();}
                respond(exchange,value==null?404:200,Map.of("order_id",id,"found",value!=null));return;
            }
            JsonNode body=JSON.readTree(exchange.getRequestBody());String id=body.path("id").asText();
            if(!id.matches("[a-z0-9-]{1,40}"))throw new IllegalArgumentException("Use a short lowercase id");
            int redisDelay=delay(body,"redis_delay_ms"),workerDelay=delay(body,"worker_delay_ms");span.setAttribute("order.id",id);
            if(body.path("fail").asBoolean(false)) {outcome="error";span.setStatus(StatusCode.ERROR,"injected application failure");event(ORDERS,"order.failed",id);respond(exchange,503,Map.of("error","injected-application-error","trace_id",span.getSpanContext().getTraceId()));return;}
            Span redis=ORDERS.tracer.spanBuilder("redis SET order").setSpanKind(SpanKind.CLIENT)
                    .setAttribute("db.system","redis").setAttribute("fault.injected_delay_ms",redisDelay).startSpan();
            try(Scope ignored=redis.makeCurrent()){Thread.sleep(redisDelay);REDIS.setex("otelbook:order:"+id,3600,body.toString());}finally{redis.end();}
            Span producer=ORDERS.tracer.spanBuilder("rocketmq publish " + TOPIC).setSpanKind(SpanKind.PRODUCER)
                    .setAttribute("messaging.system","rocketmq").setAttribute("messaging.destination.name",TOPIC).startSpan();
            try(Scope ignored=producer.makeCurrent()) {
                Message message=new Message(TOPIC,"created",id,bytes(Map.of("id",id,"worker_delay_ms",workerDelay)));
                if(body.path("propagate").asBoolean(true)) ORDERS.sdk.getPropagators().getTextMapPropagator()
                        .inject(Context.current(),message,(carrier,key,value)->carrier.putUserProperty(key,value));
                PRODUCER.send(message);
            } finally {producer.end();}
            event(ORDERS,"order.queued",id);respond(exchange,202,Map.of("order_id",id,"state","queued","trace_id",span.getSpanContext().getTraceId()));
        } catch(Exception error) {error.printStackTrace();outcome="error";span.recordException(error);span.setStatus(StatusCode.ERROR);try{respond(exchange,500,Map.of("error",error.getClass().getSimpleName()));}catch(Exception ignored){}}
        finally {ORDERS.requests.add(1,Attributes.builder().put("route","/orders").put("outcome",outcome).build());ORDERS.duration.record((System.nanoTime()-start)/1e9,Attributes.of(io.opentelemetry.api.common.AttributeKey.stringKey("route"),"/orders"),Context.root().with(span));span.end();exchange.close();}
    }
    static void gateway(HttpExchange exchange) {
        try {
            if(exchange.getRequestURI().getPath().equals("/health")){respond(exchange,200,Map.of("status","ready"));exchange.close();return;}
            if(exchange.getRequestURI().getPath().equals("/lab/events")){respond(exchange,200,List.copyOf(EVENTS));exchange.close();return;}
            if(exchange.getRequestURI().getPath().equals("/lab/flush")){GATEWAY.flush();ORDERS.flush();WORKER.flush();respond(exchange,200,Map.of("flushed",true));exchange.close();return;}
        } catch(Exception error){exchange.close();return;}
        Context parent=GATEWAY.sdk.getPropagators().getTextMapPropagator().extract(Context.root(),exchange,HTTP_GETTER);
        Span span=GATEWAY.tracer.spanBuilder("gateway " + exchange.getRequestMethod() + " /orders").setSpanKind(SpanKind.SERVER)
                .setParent(parent).setAttribute("http.route","/orders").startSpan();long start=System.nanoTime();String outcome="ok";
        try(Scope scope=span.makeCurrent()) {
            Span client=GATEWAY.tracer.spanBuilder("gateway -> orders").setSpanKind(SpanKind.CLIENT).startSpan();
            HttpResponse<String> response;
            try(Scope ignored=client.makeCurrent()) {
                byte[] body=exchange.getRequestBody().readAllBytes();
                HttpRequest.Builder request=HttpRequest.newBuilder(URI.create("http://127.0.0.1:8081"+exchange.getRequestURI()))
                        .timeout(Duration.ofSeconds(5)).header("Content-Type","application/json")
                        .method(exchange.getRequestMethod(),HttpRequest.BodyPublishers.ofByteArray(body));
                GATEWAY.sdk.getPropagators().getTextMapPropagator().inject(Context.current(),request,(carrier,key,value)->carrier.header(key,value));
                response=HTTP.send(request.build(),HttpResponse.BodyHandlers.ofString());
                if(response.statusCode()>=500)client.setStatus(StatusCode.ERROR);
            } finally {client.end();}
            if(response.statusCode()>=500){span.setStatus(StatusCode.ERROR);outcome="error";}
            JsonNode json=JSON.readTree(response.body());event(GATEWAY,"gateway.request",json.path("order_id").asText("unknown"));respond(exchange,response.statusCode(),json);
        } catch(Exception error){span.recordException(error);span.setStatus(StatusCode.ERROR);outcome="error";try{respond(exchange,502,Map.of("error",error.getClass().getSimpleName()));}catch(Exception ignored){}}
        finally {GATEWAY.requests.add(1,Attributes.builder().put("route","/orders").put("outcome",outcome).build());GATEWAY.duration.record((System.nanoTime()-start)/1e9,Attributes.of(io.opentelemetry.api.common.AttributeKey.stringKey("route"),"/orders"),Context.root().with(span));span.end();exchange.close();}
    }
    public static void main(String[] args) throws Exception {
        String namesrv=System.getenv().getOrDefault("NAMESRV_ADDR","127.0.0.1:9876");REDIS.ping();
        PRODUCER.setNamesrvAddr(namesrv);PRODUCER.setSendMsgTimeout(10000);PRODUCER.start();
        CONSUMER.setNamesrvAddr(namesrv);CONSUMER.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        CONSUMER.setConsumeThreadMin(1);CONSUMER.setConsumeThreadMax(1);CONSUMER.subscribe(TOPIC,"*");
        CONSUMER.registerMessageListener((MessageListenerConcurrently)(messages,context)->{
            try {for(MessageExt message:messages)consume(message);return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}
            catch(Exception error){error.printStackTrace();return ConsumeConcurrentlyStatus.RECONSUME_LATER;}
        });CONSUMER.start();
        HttpServer orders=HttpServer.create(new InetSocketAddress("127.0.0.1",8081),0);orders.createContext("/orders",SystemLab::order);orders.setExecutor(Executors.newVirtualThreadPerTaskExecutor());orders.start();
        HttpServer gateway=HttpServer.create(new InetSocketAddress("127.0.0.1",8080),0);gateway.createContext("/",SystemLab::gateway);gateway.setExecutor(Executors.newVirtualThreadPerTaskExecutor());gateway.start();
        Runtime.getRuntime().addShutdownHook(new Thread(()->{gateway.stop(0);orders.stop(0);CONSUMER.shutdown();PRODUCER.shutdown();GATEWAY.close();ORDERS.close();WORKER.close();REDIS.close();}));
        System.out.println("READY http://127.0.0.1:8080");
    }
}

package book;

import io.opentelemetry.api.common.Attributes;
import io.opentelemetry.api.logs.Logger;
import io.opentelemetry.api.metrics.DoubleHistogram;
import io.opentelemetry.api.metrics.LongCounter;
import io.opentelemetry.api.trace.Tracer;
import io.opentelemetry.context.propagation.ContextPropagators;
import io.opentelemetry.api.trace.propagation.W3CTraceContextPropagator;
import io.opentelemetry.exporter.otlp.http.trace.OtlpHttpSpanExporter;
import io.opentelemetry.exporter.otlp.http.metrics.OtlpHttpMetricExporter;
import io.opentelemetry.exporter.otlp.http.logs.OtlpHttpLogRecordExporter;
import io.opentelemetry.sdk.OpenTelemetrySdk;
import io.opentelemetry.sdk.resources.Resource;
import io.opentelemetry.sdk.trace.SdkTracerProvider;
import io.opentelemetry.sdk.trace.export.BatchSpanProcessor;
import io.opentelemetry.sdk.trace.samplers.Sampler;
import io.opentelemetry.sdk.metrics.SdkMeterProvider;
import io.opentelemetry.sdk.metrics.InstrumentSelector;
import io.opentelemetry.sdk.metrics.View;
import io.opentelemetry.sdk.metrics.Aggregation;
import io.opentelemetry.sdk.metrics.export.PeriodicMetricReader;
import io.opentelemetry.sdk.logs.SdkLoggerProvider;
import io.opentelemetry.sdk.logs.export.BatchLogRecordProcessor;
import java.time.Duration;
import java.util.List;
import java.util.concurrent.TimeUnit;

/** Three real OTLP signals; no in-memory tracing substitute. */
public final class Telemetry implements AutoCloseable {
    public final String service;
    public final OpenTelemetrySdk sdk;
    public final Tracer tracer;
    public final Logger logger;
    public final LongCounter requests;
    public final DoubleHistogram duration;

    public Telemetry(String service) {
        this.service = service;
        String endpoint = System.getenv().getOrDefault("COLLECTOR_URL", "http://127.0.0.1:4318");
        Resource resource = Resource.create(Attributes.builder().put("service.name", service)
                .put("service.version", "1.0.0").put("deployment.environment.name", "book-lab").build());
        double ratio = Double.parseDouble(System.getenv().getOrDefault("SAMPLER_RATIO", "1.0"));
        SdkTracerProvider traces = SdkTracerProvider.builder().setResource(resource)
                .setSampler(Sampler.parentBased(Sampler.traceIdRatioBased(ratio)))
                .addSpanProcessor(BatchSpanProcessor.builder(OtlpHttpSpanExporter.builder()
                        .setEndpoint(endpoint + "/v1/traces").build()).setScheduleDelay(Duration.ofMillis(250)).build()).build();
        SdkMeterProvider metrics = SdkMeterProvider.builder().setResource(resource)
                .registerView(InstrumentSelector.builder().setName("lab.http.duration").build(), View.builder()
                        .setAggregation(Aggregation.explicitBucketHistogram(List.of(.005,.01,.025,.05,.1,.25,.5,1.0,2.0,5.0))).build())
                .registerMetricReader(PeriodicMetricReader.builder(OtlpHttpMetricExporter.builder()
                        .setEndpoint(endpoint + "/v1/metrics").build()).setInterval(Duration.ofSeconds(1)).build()).build();
        SdkLoggerProvider logs = SdkLoggerProvider.builder().setResource(resource)
                .addLogRecordProcessor(BatchLogRecordProcessor.builder(OtlpHttpLogRecordExporter.builder()
                        .setEndpoint(endpoint + "/v1/logs").build()).setScheduleDelay(Duration.ofMillis(250)).build()).build();
        sdk = OpenTelemetrySdk.builder().setTracerProvider(traces).setMeterProvider(metrics).setLoggerProvider(logs)
                .setPropagators(ContextPropagators.create(W3CTraceContextPropagator.getInstance())).build();
        tracer = sdk.getTracer("book.observability", "1.0.0");
        logger = sdk.getLogsBridge().get("book.observability");
        requests = sdk.getMeter("book.observability").counterBuilder("lab.requests").setUnit("{request}").build();
        duration = sdk.getMeter("book.observability").histogramBuilder("lab.http.duration").setUnit("s").build();
        for (String outcome : List.of("ok", "error")) requests.add(0, Attributes.builder().put("route","/orders").put("outcome",outcome).build());
    }
    public void flush() {
        sdk.getSdkTracerProvider().forceFlush().join(10, TimeUnit.SECONDS);
        sdk.getSdkMeterProvider().forceFlush().join(10, TimeUnit.SECONDS);
        sdk.getSdkLoggerProvider().forceFlush().join(10, TimeUnit.SECONDS);
    }
    public void close() { sdk.close(); }
}

#!/usr/bin/env python3
"""Query actual backends and preserve raw evidence. No in-memory trace oracle."""
import argparse,base64,json,time,urllib.request,urllib.error,urllib.parse
from pathlib import Path
OUT=Path('evidence');OUT.mkdir(exist_ok=True)

def request(url,body=None):
    data=json.dumps(body).encode() if body is not None else None
    try:
        with urllib.request.urlopen(urllib.request.Request(url,data=data,headers={'Content-Type':'application/json','Accept':'application/json'}),timeout=15) as response:
            raw=response.read();return response.status,json.loads(raw) if raw else None
    except urllib.error.HTTPError as error:
        raw=error.read()
        try:value=json.loads(raw)
        except Exception:value=raw.decode(errors='replace')
        return error.code,value

def wait(label,fn,timeout=75):
    start=time.monotonic();last=None
    while time.monotonic()-start<timeout:
        try:
            last=fn()
            if last:return last
        except Exception as error:last=str(error)
        time.sleep(1)
    raise AssertionError(f'{label}: timeout; last={last}')

def save(name,data):(OUT/name).write_text(json.dumps(data,ensure_ascii=False,indent=2)+'\n')
def events():return request('http://127.0.0.1:8080/lab/events')[1]
def processed(order):return next((e for e in events() if e['event']=='worker.processed' and e['order_id']==order),None)
def query(expression):
    status,result=request('http://127.0.0.1:9095/api/v1/query?'+urllib.parse.urlencode({'query':expression}))
    assert status==200,(status,result)
    return result

def identifier(value):
    if not value:return ''
    if len(value) in (16,32) and all(c in '0123456789abcdefABCDEF' for c in value):return value.lower()
    return base64.b64decode(value).hex()

def spans(trace):
    result=[]
    def walk(value,service=''):
        if isinstance(value,dict):
            if 'resource' in value:
                attrs=value['resource'].get('attributes',[])
                service=next((a['value'].get('stringValue','') for a in attrs if a['key']=='service.name'),service)
            if 'spanId' in value and 'name' in value:
                result.append(dict(name=value['name'],service=service,span_id=identifier(value['spanId']),parent=identifier(value.get('parentSpanId','')),start=int(value.get('startTimeUnixNano',0)),end=int(value.get('endTimeUnixNano',0))))
            for child in value.values():walk(child,service)
        elif isinstance(value,list):
            for child in value:walk(child,service)
    walk(trace);return result

def trace(trace_id,name,services=None,minimum=7):
    required=services or {'book-gateway','book-orders','book-worker'}
    observations=[]
    def fetch():
        status,value=request('http://127.0.0.1:3200/api/traces/'+trace_id)
        found=spans(value) if status==200 else []
        observations.append({'status':status,'spans':len(found),'services':sorted({s['service'] for s in found})})
        return value if status==200 and len(found)>=minimum and required<={s['service'] for s in found} else None
    result=wait('Tempo complete trace '+name,fetch);save('trace-arrival-'+name+'.json',observations);save('trace-'+name+'.json',result);return spans(result)

def post(order,**extra):
    status,body=request('http://127.0.0.1:8080/orders',dict(id=order,**extra))
    return status,body

def main():
    ap=argparse.ArgumentParser();ap.add_argument('--phase',choices=['full','sampled-out','collector-outage','recovered'],default='full');args=ap.parse_args()
    wait('application ready',lambda:request('http://127.0.0.1:8080/health')[0]==200)
    if args.phase=='collector-outage':
        start=time.monotonic();status,body=post('collector-outage');elapsed=time.monotonic()-start
        assert status==202 and elapsed<8,(status,body,elapsed)
        event=wait('business with collector stopped',lambda:processed('collector-outage'))
        save('collector-outage.json',{'response':body,'worker_event':event,'http_elapsed_seconds':elapsed})
        print('PASS application and real MQ consumer continue while Collector is stopped');return
    if args.phase=='recovered':
        wait('Collector health after restart',lambda:request('http://127.0.0.1:13133/')[0]==200)
        status,body=post('after-outage');assert status==202,body
        event=wait('after-outage consumer',lambda:processed('after-outage'))
        found=trace(body['trace_id'],'after-outage');assert {'book-gateway','book-orders','book-worker'}<={s['service'] for s in found}
        outage=json.loads((OUT/'collector-outage.json').read_text())
        old_code,old_body=request('http://127.0.0.1:3200/api/traces/'+outage['response']['trace_id'])
        save('collector-recovered.json',{'fresh_response':body,'fresh_worker':event,'fresh_trace_services':sorted({s['service'] for s in found}),'outage_trace_status_after_recovery':old_code})
        print('PASS fresh real traces resume after Collector restart; old-trace result recorded without promising lossless recovery');return
    if args.phase=='sampled-out':
        status,body=post('sampled-out');assert status==202,body
        event=wait('unsampled consumer',lambda:processed('sampled-out'))
        assert event['sampled'] is False,event
        request('http://127.0.0.1:8080/lab/flush');time.sleep(5)
        code,value=request('http://127.0.0.1:3200/api/traces/'+body['trace_id'])
        assert code==404,(code,value)
        result={'sampled_out_order':body,'worker_event':event,'tempo_http_status':code,'metrics':query('sum(lab_requests_total{service_name="book-gateway"})')}
        assert result['metrics']['data']['result'],result
        save('sampling-zero.json',result);print('PASS sampling=0: real consumer and metrics continue; Tempo returns 404; logs mark sampled=false');return
    wait('initial metric series',lambda:query('sum(lab_requests_total{service_name="book-gateway"})')['data']['result'])
    time.sleep(3)
    results={}
    for name,options in [('normal',{}),('redis-slow',{'redis_delay_ms':350}),('worker-slow',{'worker_delay_ms':450}),('broken-context',{'propagate':False})]:
        status,body=post(name,**options);assert status==202,(status,body)
        event=wait('worker '+name,lambda:processed(name))
        results[name]={'response':body,'worker_event':event}
        found=trace(body['trace_id'],name,{'book-gateway','book-orders'} if name=='broken-context' else None,5 if name=='broken-context' else 7)
        assert {'book-gateway','book-orders'}<={s['service'] for s in found},found
        if name!='broken-context':
            assert event['trace_id']==body['trace_id'],(name,event,body)
            assert 'book-worker' in {s['service'] for s in found},found
            producer=next(s for s in found if s['name'].startswith('rocketmq publish'))
            consumer=next(s for s in found if s['name'].startswith('rocketmq process'))
            assert consumer['parent']==producer['span_id'],(producer,consumer)
            assert any(s['span_id']==event['span_id'] for s in found),event
        else:
            assert event['trace_id']!=body['trace_id'],(event,body)
            assert 'book-worker' not in {s['service'] for s in found}
            detached=trace(event['trace_id'],'detached-worker',{'book-worker'},2)
            assert 'book-worker' in {s['service'] for s in detached}
        if name=='redis-slow':
            client=next(s for s in found if s['name']=='redis SET order')
            assert (client['end']-client['start'])/1e6>=300,client
            results[name]['injected_span_ms']=(client['end']-client['start'])/1e6
        if name=='worker-slow':
            consumer=next(s for s in found if s['name'].startswith('rocketmq process'))
            assert (consumer['end']-consumer['start'])/1e6>=400,consumer
            results[name]['consumer_span_ms']=(consumer['end']-consumer['start'])/1e6
        results[name]['spans']=found
    for number in range(7):
        status,body=post('slo-ok-'+str(number));assert status==202,body
        time.sleep(.3)
    for number in range(5):
        status,body=post('slo-error-'+str(number),fail=True);assert status==503,body
        time.sleep(1)
    request('http://127.0.0.1:8080/lab/flush')
    metrics=wait('metric counter',lambda:(lambda r:r if r['data']['result'] and float(r['data']['result'][0]['value'][1])>=16 else None)(query('sum(lab_requests_total{service_name="book-gateway"})')))
    save('metrics-requests.json',metrics)
    histogram=query('histogram_quantile(0.95, sum by (le) (rate(lab_http_duration_seconds_bucket{service_name="book-gateway"}[1m])))');save('metrics-p95.json',histogram)
    assert histogram['data']['result'],histogram
    def firing():
        code,value=request('http://127.0.0.1:9095/api/v1/alerts')
        return value if code==200 and any(a['labels'].get('alertname')=='BookFastBurn' and a['state']=='firing' for a in value['data']['alerts']) else None
    alert=wait('Prometheus fast-burn firing',firing);save('alerts.json',alert)
    metric_labels=query('lab_requests_total');save('metric-labels.json',metric_labels)
    assert all('order_id' not in row['metric'] and 'trace_id' not in row['metric'] for row in metric_labels['data']['result'])
    def exemplars():
        expression='lab_http_duration_seconds_bucket{service_name="book-gateway"}'
        code,value=request('http://127.0.0.1:9095/api/v1/query_exemplars?'+urllib.parse.urlencode({'query':expression,'start':time.time()-600,'end':time.time()}))
        return value if code==200 and value.get('data') else None
    samples=wait('real metric exemplars',exemplars);save('exemplars.json',samples)
    exemplar_ids={sample['labels']['trace_id'] for series in samples['data'] for sample in series['exemplars'] if 'trace_id' in sample['labels']}
    assert exemplar_ids, samples
    known={e['trace_id'] for e in events()};assert exemplar_ids & known,(exemplar_ids,known)
    save('events.json',events());save('results.json',{'checks':results,'passed':True,'metrics_requests':metrics,'alert':alert,'exemplar_trace_ids':sorted(exemplar_ids)})
    print('PASS actual Java/Redis/RocketMQ/OTLP/Tempo/Prometheus: parentage, logs, delays, broken-context, histogram, low-cardinality labels and firing alert')
if __name__=='__main__':main()