Repository: apache/flink
File: flink-metrics/flink-metrics-otel/src/main/java/org/apache/flink/events/otel/OpenTelemetryEventReporter.java (Java)

Code before:
    private static final Logger LOG = LoggerFactory.getLogger(OpenTelemetryEventReporter.class);
    private LogRecordExporter logRecordExporter;
    private SdkLoggerProvider loggerProvider;
    private BatchLogRecordProcessor logRecordProcessor;

    public OpenTelemetryEventReporter() {
        super();
    }

    @Override
    public void open(MetricConfig metricConfig) {
        LOG.info("Starting OpenTelemetryEventReporter");
        final String protocol =
                metricConfig.getProperty(OpenTelemetryReporterOptions.EXPORTER_PROTOCOL.key());

        if (Protocol.HTTP.name().equalsIgnoreCase(protocol)) {
            OtlpHttpLogRecordExporterBuilder builder = OtlpHttpLogRecordExporter.builder();
            tryConfigureEndpoint(metricConfig, builder::setEndpoint);
            tryConfigureTimeout(metricConfig, builder::setTimeout);
            logRecordExporter = builder.build();
        } else {
            OtlpGrpcLogRecordExporterBuilder builder = OtlpGrpcLogRecordExporter.builder();
            tryConfigureEndpoint(metricConfig, builder::setEndpoint);
            tryConfigureTimeout(metricConfig, builder::setTimeout);
            logRecordExporter = builder.build();
        }

        logRecordProcessor = BatchLogRecordProcessor.builder(logRecordExporter).build();
        loggerProvider =
                SdkLoggerProvider.builder()
                        .addLogRecordProcessor(logRecordProcessor)
                        .setResource(resource)
                        .build();
    }

    @Override
    public void close() {
        logRecordProcessor.forceFlush();
        logRecordProcessor.close();
        logRecordExporter.flush();
        logRecordExporter.close();
    }

    @Override
    public void notifyOfAddedEvent(Event event) {
        io.opentelemetry.api.logs.Logger logger = loggerProvider.get(event.getClassScope());
        LogRecordBuilder logRecordBuilder = logger.logRecordBuilder();

        logRecordBuilder.setAttribute(AttributeKey.stringKey(NAME_ATTRIBUTE), event.getName());
        event.getAttributes().forEach(setAttribute(logRecordBuilder));


Review comment (suggestion):
I think it would be a better UX to at least log a warning if we see  an unexpected protocol, instead of just silently doing grpc.

Code after:
    private static final Logger LOG = LoggerFactory.getLogger(OpenTelemetryEventReporter.class);
    private LogRecordExporter logRecordExporter;
    private SdkLoggerProvider loggerProvider;
    private BatchLogRecordProcessor logRecordProcessor;

    public OpenTelemetryEventReporter() {
        super();
    }

    @Override
    public void open(MetricConfig metricConfig) {
        LOG.info("Starting OpenTelemetryEventReporter");
        final String protocol =
                Optional.ofNullable(
                                metricConfig.getProperty(
                                        OpenTelemetryReporterOptions.EXPORTER_PROTOCOL.key()))
                        .orElse("");

        switch (protocol.toLowerCase()) {
            case "http":
                OtlpHttpLogRecordExporterBuilder httpBuilder = OtlpHttpLogRecordExporter.builder();
                tryConfigureEndpoint(metricConfig, httpBuilder::setEndpoint);
                tryConfigureTimeout(metricConfig, httpBuilder::setTimeout);
                logRecordExporter = httpBuilder.build();
                break;
            default:
                LOG.warn(
                        "Unknown protocol '{}' for OpenTelemetryEventReporter, defaulting to gRPC",
                        protocol);
            // Fall through to the "gRPC" case
            case "grpc":
                OtlpGrpcLogRecordExporterBuilder grpcBuilder = OtlpGrpcLogRecordExporter.builder();
                tryConfigureEndpoint(metricConfig, grpcBuilder::setEndpoint);
                tryConfigureTimeout(metricConfig, grpcBuilder::setTimeout);
                logRecordExporter = grpcBuilder.build();
                break;
        }

        logRecordProcessor = BatchLogRecordProcessor.builder(logRecordExporter).build();
        loggerProvider =
                SdkLoggerProvider.builder()
                        .addLogRecordProcessor(logRecordProcessor)
                        .setResource(resource)
                        .build();
    }

    @Override
    public void close() {
        logRecordProcessor.forceFlush();
        logRecordProcessor.close();
        logRecordExporter.flush();