feat(core): introduce s3stream tracing

build.gradle

@@ -960,14 +960,19 @@ project(':core') {
       exclude group: 'org.slf4j', module: '*'
       exclude group: 'net.sourceforge.argparse4j', module: '*'
     }
-    implementation libs.opentelemetryApi
+
+    implementation libs.opentelemetryJava8
+    implementation libs.opentelemetryOshi
     implementation libs.opentelemetrySdk
     implementation libs.opentelemetrySdkMetrics
     implementation libs.opentelemetryExporterLogging
-    implementation libs.opentelemetrySemconv
     implementation libs.opentelemetryExporterProm
     implementation libs.opentelemetryExporterOTLP
 
+    implementation(libs.oshi) {
+      exclude group: 'org.slf4j', module: '*'
+    }
+
     // https://github.com/netty/netty-tcnative/issues/716
     // After 2.0.48, gradle project need explicitly declare the tcnative dependencies with classifiers
     implementation 'io.netty:netty-tcnative-boringssl-static:2.0.48.Final'

core/src/main/scala/kafka/log/stream/s3/DefaultS3Client.java

@@ -38,7 +38,6 @@
 import com.automq.stream.utils.threads.S3StreamThreadPoolMonitor;
 import com.automq.stream.utils.LogContext;
 import kafka.log.stream.s3.metadata.StreamMetadataManager;
-import kafka.log.stream.s3.metrics.MetricsExporter;
 import kafka.log.stream.s3.network.ControllerRequestSender;
 import kafka.log.stream.s3.objects.ControllerObjectManager;
 import kafka.log.stream.s3.streams.ControllerStreamManager;
@@ -75,7 +74,6 @@ public class DefaultS3Client implements Client {
 
     private final AsyncNetworkBandwidthLimiter networkInboundLimiter;
     private final AsyncNetworkBandwidthLimiter networkOutboundLimiter;
-    private final MetricsExporter metricsExporter;
 
     public DefaultS3Client(BrokerServer brokerServer, KafkaConfig kafkaConfig) {
         this.config = ConfigUtils.to(kafkaConfig);
@@ -112,7 +110,6 @@ public DefaultS3Client(BrokerServer brokerServer, KafkaConfig kafkaConfig) {
         }
         S3StreamThreadPoolMonitor.config(new LogContext("ThreadPoolMonitor").logger("s3.threads.logger"), TimeUnit.SECONDS.toMillis(5));
         S3StreamThreadPoolMonitor.init();
-        this.metricsExporter = new MetricsExporter(kafkaConfig);
     }
 
     @Override
@@ -130,7 +127,6 @@ public void shutdown() {
         this.networkInboundLimiter.shutdown();
         this.networkOutboundLimiter.shutdown();
         this.requestSender.shutdown();
-        this.metricsExporter.shutdown();
         LOGGER.info("S3Client shutdown successfully");
     }
 

core/src/main/scala/kafka/log/stream/s3/telemetry/ContextUtils.java

@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.log.stream.s3.telemetry;
+
+import com.automq.stream.s3.context.AppendContext;
+import com.automq.stream.s3.context.FetchContext;
+import com.automq.stream.s3.trace.context.TraceContext;
+import io.opentelemetry.api.trace.Tracer;
+import io.opentelemetry.context.Context;
+import io.opentelemetry.sdk.OpenTelemetrySdk;
+
+public class ContextUtils {
+    public static FetchContext creaetFetchContext() {
+        return new FetchContext(createTraceContext());
+    }
+
+    public static AppendContext createAppendContext() {
+        return new AppendContext(createTraceContext());
+    }
+
+    public static TraceContext createTraceContext() {
+        OpenTelemetrySdk openTelemetrySdk = TelemetryManager.getOpenTelemetrySdk();
+        boolean isTraceEnabled = openTelemetrySdk != null && TelemetryManager.isTraceEnable();
+        Tracer tracer = null;
+        if (isTraceEnabled) {
+            tracer = openTelemetrySdk.getTracer(TelemetryConstants.TELEMETRY_SCOPE_NAME);
+        }
+        return new TraceContext(isTraceEnabled, tracer, Context.current());
+    }
+
+}

core/src/main/scala/kafka/log/stream/s3/telemetry/TelemetryConstants.java

@@ -0,0 +1,29 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.log.stream.s3.telemetry;
+
+import io.opentelemetry.api.common.AttributeKey;
+
+public class TelemetryConstants {
+    public static final String TELEMETRY_SCOPE_NAME = "automq_for_kafka";
+    public static final String KAFKA_METRICS_PREFIX = "kafka_stream_";
+    public static final AttributeKey<Long> STREAM_ID_NAME = AttributeKey.longKey("streamId");
+    public static final AttributeKey<Long> START_OFFSET_NAME = AttributeKey.longKey("startOffset");
+    public static final AttributeKey<Long> END_OFFSET_NAME = AttributeKey.longKey("endOffset");
+    public static final AttributeKey<Long> MAX_BYTES_NAME = AttributeKey.longKey("maxBytes");
+}