使用Java接收和处理OpenTelemetry数据的完整指南

import io.opentelemetry.api.metrics.Meter;
import io.opentelemetry.exporter.otlp.metrics.OtlpGrpcMetricExporter;
import io.opentelemetry.exporter.otlp.metrics.OtlpMetricExporterBuilder;
import io.opentelemetry.sdk.metrics.SdkMeterProvider;
import io.opentelemetry.sdk.metrics.export.IntervalMetricReader;
import io.opentelemetry.sdk.metrics.export.IntervalMetricReaderBuilder;
import io.opentelemetry.sdk.metrics.export.MetricExporter;
import io.opentelemetry.sdk.metrics.export.MetricExporterBuilder;
import io.opentelemetry.sdk.metrics.export.MetricProducer;

public class OTLPReceiverExample {

    public static void main(String[] args) {
        // 创建 OTLP Metric Exporter
        MetricExporterBuilder exporterBuilder = OtlpMetricExporter.builder();
        MetricExporter exporter = exporterBuilder.build();

        // 创建 Meter 和 Metric Producer
        SdkMeterProvider meterProvider = SdkMeterProvider.builder().build();
        Meter meter = meterProvider.get("OTLPReceiverExample");
        MetricProducer metricProducer = meterProvider;

        // 创建 Interval Metric Reader
        IntervalMetricReaderBuilder readerBuilder = IntervalMetricReader.builder();
        IntervalMetricReader reader = readerBuilder.setMetricExporter(exporter)
                                                    .setMetricProducer(metricProducer)
                                                    .build();

        // 启动 Interval Metric Reader
        reader.start();
        
        // 等待一段时间，接收和处理 OTLP 数据
        try {
            Thread.sleep(60 * 1000); // 等待 60 秒
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        // 关闭 Interval Metric Reader
        reader.stop();
    }
}

@PostMapping(value = "/log-otlp", produces = "application/x-protobuf", consumes = "application/x-protobuf")
    public ExportLogsServiceResponse insertOtlpLog(
        @RequestBody ExportLogsServiceRequest exportLogsServiceRequest) {
        log.debug("insertOtlpLogs request is {}", JsonFormat.printer().print(exportLogsServiceRequest));
        return insertService.insert(exportLogsServiceRequest);
    }

import java.util.List;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;
import org.springframework.http.converter.HttpMessageConverter;
import org.springframework.http.converter.json.MappingJackson2HttpMessageConverter;
import org.springframework.http.converter.protobuf.ProtobufHttpMessageConverter;
import org.springframework.web.servlet.config.annotation.InterceptorRegistry;
import org.springframework.web.servlet.config.annotation.WebMvcConfigurer;


@Configuration
public class WebMvcConfig implements WebMvcConfigurer {

    @Autowired
    AuthInterceptor authInterceptor;

    @Override
    public void configureMessageConverters(List<HttpMessageConverter<?>> converters) {
        converters.add(new ProtobufHttpMessageConverter());
        // 如果同时支持 JSON，可以添加 MappingJackson2HttpMessageConverter
        converters.add(new MappingJackson2HttpMessageConverter());
    }

    @Override
    public void addInterceptors(InterceptorRegistry registry) {
        registry.addInterceptor(authInterceptor).addPathPatterns("/**");
    }
}

package com.darkraven.interceptor.handler;

import static io.opentelemetry.proto.trace.v1.ResourceSpans.newBuilder;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.HttpURLConnection;
import java.net.URL;
import java.util.UUID;

import io.opentelemetry.proto.collector.trace.v1.ExportTraceServiceRequest;
import io.opentelemetry.proto.common.v1.AnyValue;
import io.opentelemetry.proto.common.v1.KeyValue;
import io.opentelemetry.proto.resource.v1.Resource;
import io.opentelemetry.proto.trace.v1.ScopeSpans;
import io.opentelemetry.proto.trace.v1.Span;
import io.opentelemetry.proto.trace.v1.Status;
import okhttp3.MediaType;
import okhttp3.OkHttpClient;
import okhttp3.Request;
import okhttp3.RequestBody;
import okhttp3.Response;

import com.google.protobuf.ByteString;


public class HttpDemo {

    public static void main(String[] args) throws InterruptedException, IOException {
        int i = 0;
        while (true) {
            sendData();
            i++;
            System.out.println("这是执行的第" + i + "次");
            Thread.sleep(5000L);
        }

    }

    private static void sendData() throws IOException {

        Span span = buildSpan();

        // 创建 ExportTraceServiceRequest
        ExportTraceServiceRequest requestData = ExportTraceServiceRequest.newBuilder()
            .addResourceSpans(newBuilder()
                .setResource(Resource.newBuilder()
                    .addAttributes(KeyValue.newBuilder()
                        .setKey("service.name")
                        .setValue(AnyValue.newBuilder().setStringValue("darkraven").build())
                        .build())
                    .build())
                .addScopeSpans(ScopeSpans.newBuilder().addSpans(span)))
            .build();
        // 创建 OkHttpClient 实例
        OkHttpClient client = new OkHttpClient();

        // 替换为你的接口URL
        String apiUrl = "http://localhost:32167/api/v1/insert/trace-otlp";

        // 构造 RequestBody，设置为 Protobuf 格式
        RequestBody body = RequestBody.create(requestData.toByteArray(), MediaType.get("application/x-protobuf"));

        // 构造请求对象
        Request request = new Request.Builder()
            .url(apiUrl)
            .post(body)
            .addHeader("apikey", "40c996be-cf3a-4ba0-9697-2feef2a76f0e")
            .build();

        // 发送请求并获取响应
        try (Response response = client.newCall(request).execute()) {
            if (!response.isSuccessful()) {
                throw new RuntimeException("Unexpected response code: " + response);
            }

            // 处理响应内容
            System.out.println("Response code: " + response.code());
            System.out.println("Response body: " + response.body().string());
        }

    }


    public static Span buildSpan() {
        io.opentelemetry.proto.trace.v1.Span span = io.opentelemetry.proto.trace.v1.Span.newBuilder()
            .setTraceId(ByteString.fromHex("1234567890abcdef1234567890abcdef"))
            .setSpanId(ByteString.fromHex("1234567890abcdef"))
            .setParentSpanId(ByteString.fromHex("abcdef1234567890"))
            .setName("example-span")
            .setKind(io.opentelemetry.proto.trace.v1.Span.SpanKind.SPAN_KIND_CLIENT)
            .setStartTimeUnixNano(System.currentTimeMillis() * 1_000_000)
            .setEndTimeUnixNano((System.currentTimeMillis() + 1000) * 1_000_000)
            .setStatus(Status.newBuilder()
                .setCode(Status.StatusCode.STATUS_CODE_OK)
                .setMessage("example-status"))
            .addAttributes(KeyValue.newBuilder()
                .setKey("attribute_key")
                .setValue(AnyValue.newBuilder().setStringValue("attribute_value")))
            .build();
        return span;
    }

    public static String generateApiKey() {
        return UUID.randomUUID().toString();
    }
}