feat: add unit tests to otel-integration

Signed-off-by: chenhuan <[email protected]>

Author: chenhuansome

opengemini-client-api/src/main/java/io/opengemini/client/api/Query.java

@@ -16,7 +16,6 @@
 
 package io.opengemini.client.api;
 
-import io.opentelemetry.api.trace.Span;
 import lombok.AllArgsConstructor;
 import lombok.Getter;
 import lombok.Setter;
@@ -74,4 +73,4 @@ public void setAttribute(String key, Object value) {
     public Object getAttribute(String key) {
         return attributes.get(key);
     }
-}
+}

opengemini-client-api/src/main/java/io/opengemini/client/api/Write.java

@@ -49,4 +49,4 @@ public void setAttribute(String key, Object value) {
     public Object getAttribute(String key) {
         return attributes.get(key);
     }
-}
+}

opengemini-client-common/src/main/java/io/opengemini/client/common/BaseClient.java

@@ -162,4 +162,4 @@ protected String getQueryUrl(Query query) {
     public void close() throws IOException {
         scheduler.ifPresent(ExecutorService::shutdown);
     }
-}
+}

opengemini-client/src/main/java/io/opengemini/client/impl/OpenGeminiClient.java

@@ -38,18 +38,7 @@
 
 public class OpenGeminiClient extends BaseClient implements OpenGeminiAsyncClient {
     private final List<Interceptor> interceptors = new ArrayList<>();
-
-    /**
-     * Add interceptors to the client.
-     */
-    public void addInterceptors(Interceptor... interceptors) {
-        Collections.addAll(this.interceptors, interceptors);
-    }
-
-
-
     protected final Configuration conf;
-
     private final HttpClient client;
 
     public OpenGeminiClient(@NotNull Configuration conf) {
@@ -68,6 +57,10 @@ public OpenGeminiClient(@NotNull Configuration conf) {
         this.client = HttpClientFactory.createHttpClient(httpConfig);
     }
 
+    public void addInterceptors(Interceptor... interceptors) {
+        Collections.addAll(this.interceptors, interceptors);
+    }
+
     /**
      * {@inheritDoc}
      */
@@ -163,7 +156,7 @@ public CompletableFuture<Void> write(String database, String retentionPolicy, Po
         if (StringUtils.isEmpty(body)) {
             return CompletableFuture.completedFuture(null);
         }
-        return executeWrite(database, retentionPolicy, body );
+        return executeWrite(database, retentionPolicy, body);
     }
 
     @Override
@@ -201,25 +194,20 @@ public CompletableFuture<Pong> ping() {
      */
     public CompletableFuture<QueryResult> executeQuery(Query query) {
         String queryUrl = getQueryUrl(query);
-
-        // 执行所有queryBefore拦截器
         CompletableFuture<Void> beforeFutures = CompletableFuture.allOf(
                 interceptors.stream()
                         .map(interceptor -> interceptor.queryBefore(query))
                         .toArray(CompletableFuture[]::new)
         );
 
-        return beforeFutures.thenCompose(voidResult -> {
-            return executeHttpQuery(query).thenCompose(response -> {
-                // 执行所有queryAfter拦截器
-                CompletableFuture<Void> afterFutures = CompletableFuture.allOf(
-                        interceptors.stream()
-                                .map(interceptor -> interceptor.queryAfter(query, response))
-                                .toArray(CompletableFuture[]::new)
-                );
-                return afterFutures.thenCompose(voidResult2 -> convertResponse(response, QueryResult.class));
-            });
-        });
+        return beforeFutures.thenCompose(voidResult -> executeHttpQuery(query).thenCompose(response -> {
+            CompletableFuture<Void> afterFutures = CompletableFuture.allOf(
+                    interceptors.stream()
+                            .map(interceptor -> interceptor.queryAfter(query, response))
+                            .toArray(CompletableFuture[]::new)
+            );
+            return afterFutures.thenCompose(voidResult2 -> convertResponse(response, QueryResult.class));
+        }));
     }
 
     /**
@@ -244,28 +232,28 @@ public CompletableFuture<Void> executeWrite(String database, String retentionPol
         Write write = new Write(
                 database,
                 retentionPolicy,
-                "default_measurement", // Default measurement name
+                "default_measurement",
                 lineProtocol,
-                "ns" // Default precision
+                "ns"
         );
 
-        // Execute all writeBefore interceptors
         CompletableFuture<Void> beforeFutures = CompletableFuture.allOf(
                 interceptors.stream()
                         .map(interceptor -> interceptor.writeBefore(write))
                         .toArray(CompletableFuture[]::new)
         );
 
-        return beforeFutures.thenCompose(voidResult -> {
-            return executeHttpWrite(write).thenCompose(response -> { // response 是 io.github.openfacade.http.HttpResponse
-                CompletableFuture<Void> afterFutures = CompletableFuture.allOf(
-                        interceptors.stream()
-                                .map(interceptor -> interceptor.writeAfter(write, response)) // 传递正确的类型
-                                .toArray(CompletableFuture[]::new)
-                );
-                return afterFutures.thenCompose(voidResult2 -> convertResponse(response, Void.class));
-            });
-        });
+        return beforeFutures.thenCompose(voidResult ->
+                executeHttpWrite(write).thenCompose(response -> {
+                    CompletableFuture<Void> afterFutures = CompletableFuture.allOf(
+                            interceptors.stream()
+                                    .map(interceptor -> interceptor.writeAfter(write, response))
+                                    .toArray(CompletableFuture[]::new)
+                    );
+                    return afterFutures.thenCompose(voidResult2 ->
+                            convertResponse(response, Void.class));
+                })
+        );
     }
 
     /**
@@ -317,19 +305,13 @@ public String toString() {
         return "OpenGeminiClient{" + "httpEngine=" + conf.getHttpConfig().engine() + '}';
     }
 
-    /**
-     * 执行 HTTP 查询请求
-     */
     private CompletableFuture<HttpResponse> executeHttpQuery(Query query) {
         String queryUrl = getQueryUrl(query);
         return get(queryUrl);
     }
 
-    /**
-     * 执行 HTTP 写入请求
-     */
     private CompletableFuture<HttpResponse> executeHttpWrite(Write write) {
         String writeUrl = getWriteUrl(write.getDatabase(), write.getRetentionPolicy());
         return post(writeUrl, write.getLineProtocol());
     }
-}
+}

opengemini-client/src/main/java/io/opengemini/client/impl/OpenGeminiClientFactory.java

@@ -67,4 +67,4 @@ public static OpenGeminiClient createClientWithInterceptors(Configuration config
         client.addInterceptors(interceptors.toArray(new Interceptor[0]));
         return client;
     }
-}
+}

opengemini-client/src/main/java/io/opengemini/client/impl/OpenTelemetryConfig.java

@@ -30,7 +30,6 @@
  * Configuration for OpenTelemetry tracing.
  */
 public class OpenTelemetryConfig {
-
     private static volatile OpenTelemetry openTelemetry;
     private static volatile Tracer tracer;
 
@@ -73,4 +72,4 @@ public static Tracer getTracer() {
         }
         return tracer;
     }
-}
+}

opengemini-client/src/main/java/io/opengemini/client/interceptor/Interceptor.java

@@ -61,4 +61,4 @@ public interface Interceptor {
      * @return a CompletableFuture that completes when the interceptor logic is done
      */
     CompletableFuture<Void> writeAfter(Write write, HttpResponse response);
-}
+}

opengemini-client/src/main/java/io/opengemini/client/interceptor/OtelInterceptor.java

@@ -137,4 +137,4 @@ public CompletableFuture<Void> writeAfter(Write write, HttpResponse response) {
             }
         });
     }
-}
+}

opengemini-client/src/test/java/io/opengemini/client/interceptor/TracingIntegrationTest.java

@@ -0,0 +1,189 @@
+/*
+ * Copyright 2024 openGemini Authors
+ *
+ * Licensed 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 io.opengemini.client.interceptor;
+
+import io.github.openfacade.http.HttpClientConfig;
+import io.opengemini.client.api.*;
+import io.opengemini.client.impl.OpenGeminiClient;
+import io.opengemini.client.impl.OpenGeminiClientFactory;
+import io.opengemini.client.impl.OpenTelemetryConfig;
+import lombok.Getter;
+import lombok.Setter;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Duration;
+import java.util.Collections;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+
+/**
+ * Example demonstrating OpenGemini client usage with interceptors.
+ */
+
+@Getter
+public class TracingIntegrationTest {
+    @Setter
+    private String database;
+    @Setter
+    private String retentionPolicy;
+    @Setter
+    private String lineProtocol;
+    private static final Logger LOG = LoggerFactory.getLogger(TracingIntegrationTest.class);
+
+    private OpenGeminiClient openGeminiClient;
+
+    @BeforeEach
+    void setUp() {
+        HttpClientConfig httpConfig = new HttpClientConfig.Builder()
+                .connectTimeout(Duration.ofSeconds(3))
+                .timeout(Duration.ofSeconds(3))
+                .build();
+        Configuration configuration = Configuration.builder()
+                .addresses(Collections.singletonList(new Address("127.0.0.1", 8086)))
+                .httpConfig(httpConfig)
+                .authConfig(new AuthConfig(AuthType.PASSWORD, "test", "testPwd123@".toCharArray(), null))
+                .gzipEnabled(false)
+                .build();
+        this.openGeminiClient = new OpenGeminiClient(configuration);
+    }
+
+    @Test
+    void testClientCreation() {
+        Configuration config = new Configuration();
+        config.setAddresses(java.util.Collections.singletonList(new io.opengemini.client.api.Address("localhost", 8086)));
+        if (config.getHttpConfig() == null) {
+            config.setHttpConfig(new HttpClientConfig.Builder().build());
+        }
+
+        assertDoesNotThrow(() -> {
+            OpenGeminiClient client = OpenGeminiClientFactory.createClientWithInterceptors(config);
+            assertNotNull(client, "OpenGeminiClient should be created successfully");
+            client.close();
+        }, "Client creation should not throw an exception");
+    }
+
+    @Test
+    void testDatabaseCreation() {
+        Configuration config = new Configuration();
+        config.setAddresses(java.util.Collections.singletonList(new io.opengemini.client.api.Address("localhost", 8086)));
+        if (config.getHttpConfig() == null) {
+            config.setHttpConfig(new HttpClientConfig.Builder().build());
+        }
+
+        assertDoesNotThrow(() -> {
+            try (OpenGeminiClient client = OpenGeminiClientFactory.createClientWithInterceptors(config)) {
+                Query createDbQuery = new Query("CREATE DATABASE test_db");
+                client.query(createDbQuery).get(10, TimeUnit.SECONDS);
+            }
+        }, "Database creation should not throw an exception");
+    }
+
+    @Test
+    void testQueryOperation() {
+        Configuration config = new Configuration();
+        config.setAddresses(java.util.Collections.singletonList(new Address("localhost", 8086)));
+        if (config.getHttpConfig() == null) {
+            config.setHttpConfig(new HttpClientConfig.Builder().build());
+        }
+
+        assertDoesNotThrow(() -> {
+            try (OpenGeminiClient client = OpenGeminiClientFactory.createClientWithInterceptors(config)) {
+                Query createDbQuery = new Query("CREATE DATABASE test_db");
+                client.query(createDbQuery).get(10, TimeUnit.SECONDS);
+
+                Query showDbQuery = new Query("SHOW DATABASES");
+                QueryResult result = client.query(showDbQuery).get(10, TimeUnit.SECONDS);
+                assertNotNull(result, "Query result should not be null");
+            }
+        }, "Query operation should not throw an exception");
+    }
+
+    @BeforeAll
+    static void initializeTracing() {
+        OpenTelemetryConfig.initialize();
+    }
+
+    @Test
+    void testWriteOperation() {
+        Configuration config = new Configuration();
+        config.setAddresses(java.util.Collections.singletonList(
+                new Address("localhost", 8086)));
+
+        if (config.getHttpConfig() == null) {
+            config.setHttpConfig(new HttpClientConfig.Builder().build());
+        }
+
+        assertDoesNotThrow(() -> {
+            try (OpenGeminiClient client = OpenGeminiClientFactory.createClientWithInterceptors(config)) {
+                Query createDbQuery = new Query("CREATE DATABASE test_db");
+                client.query(createDbQuery).get(10, TimeUnit.SECONDS);
+
+                Thread.sleep(1000);
+
+                Write write = new Write(
+                        "test_db",
+                        "autogen",
+                        "temperature",
+                        "temperature,location=room1 value=25.5 " + System.currentTimeMillis(),
+                        "ns"
+                );
+
+                client.executeWrite(
+                        write.getDatabase(),
+                        write.getRetentionPolicy(),
+                        write.getLineProtocol()
+                ).get(10, TimeUnit.SECONDS);
+            }
+        }, "Write operation should not throw an exception");
+    }
+
+    @Test
+    void testTracingIntegration() throws ExecutionException, InterruptedException {
+        String databaseTestName = "tracing_test_db";
+        CompletableFuture<Void> createdb = openGeminiClient.createDatabase(databaseTestName);
+        createdb.get();
+
+        assertDoesNotThrow(() -> {
+
+            Write write = new Write(
+                    "tracing_test_db",
+                    "autogen",
+                    "tracing_measurement",
+                    "tracing_measurement,tag=test value=8 " + System.currentTimeMillis(),
+                    "ns"
+            );
+
+            openGeminiClient.executeWrite(
+                    write.getDatabase(),
+                    write.getRetentionPolicy(),
+                    write.getLineProtocol()
+            ).get(10, TimeUnit.SECONDS);
+
+            Query query = new Query("SELECT * FROM tracing_measurement");
+            openGeminiClient.query(query).get(10, TimeUnit.SECONDS);
+
+        }, "Tracing integration should not throw an exception");
+    }
+}