| 
 | 1 | +/*  | 
 | 2 | + * Copyright The OpenTelemetry Authors  | 
 | 3 | + * SPDX-License-Identifier: Apache-2.0  | 
 | 4 | + */  | 
 | 5 | + | 
 | 6 | +package io.opentelemetry.javaagent.instrumentation.hibernate.reactive.v2_0;  | 
 | 7 | + | 
 | 8 | +import static io.opentelemetry.instrumentation.api.internal.SemconvStability.emitStableDatabaseSemconv;  | 
 | 9 | +import static io.opentelemetry.instrumentation.testing.junit.db.SemconvStabilityUtil.maybeStable;  | 
 | 10 | +import static io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions.equalTo;  | 
 | 11 | +import static io.opentelemetry.semconv.ServerAttributes.SERVER_ADDRESS;  | 
 | 12 | +import static io.opentelemetry.semconv.ServerAttributes.SERVER_PORT;  | 
 | 13 | +import static io.opentelemetry.semconv.incubating.DbIncubatingAttributes.DB_NAME;  | 
 | 14 | +import static io.opentelemetry.semconv.incubating.DbIncubatingAttributes.DB_OPERATION;  | 
 | 15 | +import static io.opentelemetry.semconv.incubating.DbIncubatingAttributes.DB_SQL_TABLE;  | 
 | 16 | +import static io.opentelemetry.semconv.incubating.DbIncubatingAttributes.DB_STATEMENT;  | 
 | 17 | +import static io.opentelemetry.semconv.incubating.DbIncubatingAttributes.DB_USER;  | 
 | 18 | + | 
 | 19 | +import io.opentelemetry.api.trace.Span;  | 
 | 20 | +import io.opentelemetry.api.trace.SpanKind;  | 
 | 21 | +import io.opentelemetry.instrumentation.testing.junit.AgentInstrumentationExtension;  | 
 | 22 | +import io.opentelemetry.instrumentation.testing.junit.InstrumentationExtension;  | 
 | 23 | +import io.vertx.core.Vertx;  | 
 | 24 | +import jakarta.persistence.EntityManagerFactory;  | 
 | 25 | +import java.time.Duration;  | 
 | 26 | +import java.util.concurrent.CompletableFuture;  | 
 | 27 | +import java.util.concurrent.TimeUnit;  | 
 | 28 | +import org.hibernate.reactive.mutiny.Mutiny;  | 
 | 29 | +import org.hibernate.reactive.stage.Stage;  | 
 | 30 | +import org.junit.jupiter.api.AfterAll;  | 
 | 31 | +import org.junit.jupiter.api.BeforeAll;  | 
 | 32 | +import org.junit.jupiter.api.Test;  | 
 | 33 | +import org.junit.jupiter.api.TestInstance;  | 
 | 34 | +import org.junit.jupiter.api.extension.RegisterExtension;  | 
 | 35 | +import org.slf4j.Logger;  | 
 | 36 | +import org.slf4j.LoggerFactory;  | 
 | 37 | +import org.testcontainers.containers.GenericContainer;  | 
 | 38 | +import org.testcontainers.containers.output.Slf4jLogConsumer;  | 
 | 39 | + | 
 | 40 | +@SuppressWarnings("InterruptedExceptionSwallowed")  | 
 | 41 | +@TestInstance(TestInstance.Lifecycle.PER_CLASS)  | 
 | 42 | +public abstract class AbstractHibernateReactiveTest {  | 
 | 43 | +  private static final Logger logger = LoggerFactory.getLogger(AbstractHibernateReactiveTest.class);  | 
 | 44 | + | 
 | 45 | +  private static final String USER_DB = "SA";  | 
 | 46 | +  private static final String PW_DB = "password123";  | 
 | 47 | +  private static final String DB = "tempdb";  | 
 | 48 | + | 
 | 49 | +  @RegisterExtension  | 
 | 50 | +  protected static final InstrumentationExtension testing = AgentInstrumentationExtension.create();  | 
 | 51 | + | 
 | 52 | +  protected final Vertx vertx = Vertx.vertx();  | 
 | 53 | +  private GenericContainer<?> container;  | 
 | 54 | +  private String host;  | 
 | 55 | +  private int port;  | 
 | 56 | +  private EntityManagerFactory entityManagerFactory;  | 
 | 57 | +  private Mutiny.SessionFactory mutinySessionFactory;  | 
 | 58 | +  private Stage.SessionFactory stageSessionFactory;  | 
 | 59 | + | 
 | 60 | +  protected abstract EntityManagerFactory createEntityManagerFactory() throws Exception;  | 
 | 61 | + | 
 | 62 | +  @BeforeAll  | 
 | 63 | +  void setUp() throws Exception {  | 
 | 64 | +    container =  | 
 | 65 | +        new GenericContainer<>("postgres:9.6.8")  | 
 | 66 | +            .withEnv("POSTGRES_USER", USER_DB)  | 
 | 67 | +            .withEnv("POSTGRES_PASSWORD", PW_DB)  | 
 | 68 | +            .withEnv("POSTGRES_DB", DB)  | 
 | 69 | +            .withExposedPorts(5432)  | 
 | 70 | +            .withLogConsumer(new Slf4jLogConsumer(logger))  | 
 | 71 | +            .withStartupTimeout(Duration.ofMinutes(2));  | 
 | 72 | +    container.start();  | 
 | 73 | + | 
 | 74 | +    host = container.getHost();  | 
 | 75 | +    port = container.getMappedPort(5432);  | 
 | 76 | +    System.setProperty("db.host", host);  | 
 | 77 | +    System.setProperty("db.port", String.valueOf(port));  | 
 | 78 | + | 
 | 79 | +    entityManagerFactory = createEntityManagerFactory();  | 
 | 80 | + | 
 | 81 | +    Value value = new Value("name");  | 
 | 82 | +    value.setId(1L);  | 
 | 83 | + | 
 | 84 | +    mutinySessionFactory = entityManagerFactory.unwrap(Mutiny.SessionFactory.class);  | 
 | 85 | +    stageSessionFactory = entityManagerFactory.unwrap(Stage.SessionFactory.class);  | 
 | 86 | + | 
 | 87 | +    mutinySessionFactory  | 
 | 88 | +        .withTransaction((session, tx) -> session.merge(value))  | 
 | 89 | +        .await()  | 
 | 90 | +        .atMost(Duration.ofSeconds(30));  | 
 | 91 | +  }  | 
 | 92 | + | 
 | 93 | +  @AfterAll  | 
 | 94 | +  void cleanUp() {  | 
 | 95 | +    if (entityManagerFactory != null) {  | 
 | 96 | +      entityManagerFactory.close();  | 
 | 97 | +    }  | 
 | 98 | +    if (mutinySessionFactory != null) {  | 
 | 99 | +      mutinySessionFactory.close();  | 
 | 100 | +    }  | 
 | 101 | +    if (stageSessionFactory != null) {  | 
 | 102 | +      stageSessionFactory.close();  | 
 | 103 | +    }  | 
 | 104 | +    vertx.close();  | 
 | 105 | +    container.stop();  | 
 | 106 | +  }  | 
 | 107 | + | 
 | 108 | +  @Test  | 
 | 109 | +  void testMutiny() {  | 
 | 110 | +    testing.runWithSpan(  | 
 | 111 | +        "parent",  | 
 | 112 | +        () -> {  | 
 | 113 | +          mutinySessionFactory  | 
 | 114 | +              .withSession(  | 
 | 115 | +                  session -> {  | 
 | 116 | +                    if (!Span.current().getSpanContext().isValid()) {  | 
 | 117 | +                      throw new IllegalStateException("missing parent span");  | 
 | 118 | +                    }  | 
 | 119 | + | 
 | 120 | +                    return session  | 
 | 121 | +                        .find(Value.class, 1L)  | 
 | 122 | +                        .invoke(value -> testing.runWithSpan("callback", () -> {}));  | 
 | 123 | +                  })  | 
 | 124 | +              .await()  | 
 | 125 | +              .atMost(Duration.ofSeconds(30));  | 
 | 126 | +        });  | 
 | 127 | + | 
 | 128 | +    assertTrace();  | 
 | 129 | +  }  | 
 | 130 | + | 
 | 131 | +  @Test  | 
 | 132 | +  void testStage() throws Exception {  | 
 | 133 | +    testing  | 
 | 134 | +        .runWithSpan(  | 
 | 135 | +            "parent",  | 
 | 136 | +            () ->  | 
 | 137 | +                stageSessionFactory  | 
 | 138 | +                    .withSession(  | 
 | 139 | +                        session -> {  | 
 | 140 | +                          if (!Span.current().getSpanContext().isValid()) {  | 
 | 141 | +                            throw new IllegalStateException("missing parent span");  | 
 | 142 | +                          }  | 
 | 143 | + | 
 | 144 | +                          return session  | 
 | 145 | +                              .find(Value.class, 1L)  | 
 | 146 | +                              .thenAccept(value -> testing.runWithSpan("callback", () -> {}));  | 
 | 147 | +                        })  | 
 | 148 | +                    .toCompletableFuture())  | 
 | 149 | +        .get(30, TimeUnit.SECONDS);  | 
 | 150 | + | 
 | 151 | +    assertTrace();  | 
 | 152 | +  }  | 
 | 153 | + | 
 | 154 | +  @Test  | 
 | 155 | +  void testStageWithStatelessSession() throws Exception {  | 
 | 156 | +    testing  | 
 | 157 | +        .runWithSpan(  | 
 | 158 | +            "parent",  | 
 | 159 | +            () ->  | 
 | 160 | +                stageSessionFactory  | 
 | 161 | +                    .withStatelessSession(  | 
 | 162 | +                        session -> {  | 
 | 163 | +                          if (!Span.current().getSpanContext().isValid()) {  | 
 | 164 | +                            throw new IllegalStateException("missing parent span");  | 
 | 165 | +                          }  | 
 | 166 | + | 
 | 167 | +                          return session  | 
 | 168 | +                              .get(Value.class, 1L)  | 
 | 169 | +                              .thenAccept(value -> testing.runWithSpan("callback", () -> {}));  | 
 | 170 | +                        })  | 
 | 171 | +                    .toCompletableFuture())  | 
 | 172 | +        .get(30, TimeUnit.SECONDS);  | 
 | 173 | + | 
 | 174 | +    assertTrace();  | 
 | 175 | +  }  | 
 | 176 | + | 
 | 177 | +  @Test  | 
 | 178 | +  void testStageSessionWithTransaction() throws Exception {  | 
 | 179 | +    testing  | 
 | 180 | +        .runWithSpan(  | 
 | 181 | +            "parent",  | 
 | 182 | +            () ->  | 
 | 183 | +                stageSessionFactory  | 
 | 184 | +                    .withSession(  | 
 | 185 | +                        session -> {  | 
 | 186 | +                          if (!Span.current().getSpanContext().isValid()) {  | 
 | 187 | +                            throw new IllegalStateException("missing parent span");  | 
 | 188 | +                          }  | 
 | 189 | + | 
 | 190 | +                          return session  | 
 | 191 | +                              .withTransaction(transaction -> session.find(Value.class, 1L))  | 
 | 192 | +                              .thenAccept(value -> testing.runWithSpan("callback", () -> {}));  | 
 | 193 | +                        })  | 
 | 194 | +                    .toCompletableFuture())  | 
 | 195 | +        .get(30, TimeUnit.SECONDS);  | 
 | 196 | + | 
 | 197 | +    assertTrace();  | 
 | 198 | +  }  | 
 | 199 | + | 
 | 200 | +  @Test  | 
 | 201 | +  void testStageStatelessSessionWithTransaction() throws Exception {  | 
 | 202 | +    testing  | 
 | 203 | +        .runWithSpan(  | 
 | 204 | +            "parent",  | 
 | 205 | +            () ->  | 
 | 206 | +                stageSessionFactory  | 
 | 207 | +                    .withStatelessSession(  | 
 | 208 | +                        session -> {  | 
 | 209 | +                          if (!Span.current().getSpanContext().isValid()) {  | 
 | 210 | +                            throw new IllegalStateException("missing parent span");  | 
 | 211 | +                          }  | 
 | 212 | + | 
 | 213 | +                          return session  | 
 | 214 | +                              .withTransaction(transaction -> session.get(Value.class, 1L))  | 
 | 215 | +                              .thenAccept(value -> testing.runWithSpan("callback", () -> {}));  | 
 | 216 | +                        })  | 
 | 217 | +                    .toCompletableFuture())  | 
 | 218 | +        .get(30, TimeUnit.SECONDS);  | 
 | 219 | + | 
 | 220 | +    assertTrace();  | 
 | 221 | +  }  | 
 | 222 | + | 
 | 223 | +  @Test  | 
 | 224 | +  void testStageOpenSession() throws Exception {  | 
 | 225 | +    CompletableFuture<Object> result = new CompletableFuture<>();  | 
 | 226 | +    testing.runWithSpan(  | 
 | 227 | +        "parent",  | 
 | 228 | +        () ->  | 
 | 229 | +            runWithVertx(  | 
 | 230 | +                () ->  | 
 | 231 | +                    stageSessionFactory  | 
 | 232 | +                        .openSession()  | 
 | 233 | +                        .thenApply(  | 
 | 234 | +                            session -> {  | 
 | 235 | +                              if (!Span.current().getSpanContext().isValid()) {  | 
 | 236 | +                                throw new IllegalStateException("missing parent span");  | 
 | 237 | +                              }  | 
 | 238 | + | 
 | 239 | +                              return session  | 
 | 240 | +                                  .find(Value.class, 1L)  | 
 | 241 | +                                  .thenAccept(value -> testing.runWithSpan("callback", () -> {}));  | 
 | 242 | +                            })  | 
 | 243 | +                        .whenComplete((value, throwable) -> complete(result, value, throwable))));  | 
 | 244 | +    result.get(30, TimeUnit.SECONDS);  | 
 | 245 | + | 
 | 246 | +    assertTrace();  | 
 | 247 | +  }  | 
 | 248 | + | 
 | 249 | +  @Test  | 
 | 250 | +  void testStageOpenStatelessSession() throws Exception {  | 
 | 251 | +    CompletableFuture<Object> result = new CompletableFuture<>();  | 
 | 252 | +    testing.runWithSpan(  | 
 | 253 | +        "parent",  | 
 | 254 | +        () ->  | 
 | 255 | +            runWithVertx(  | 
 | 256 | +                () ->  | 
 | 257 | +                    stageSessionFactory  | 
 | 258 | +                        .openStatelessSession()  | 
 | 259 | +                        .thenApply(  | 
 | 260 | +                            session -> {  | 
 | 261 | +                              if (!Span.current().getSpanContext().isValid()) {  | 
 | 262 | +                                throw new IllegalStateException("missing parent span");  | 
 | 263 | +                              }  | 
 | 264 | + | 
 | 265 | +                              return session  | 
 | 266 | +                                  .get(Value.class, 1L)  | 
 | 267 | +                                  .thenAccept(value -> testing.runWithSpan("callback", () -> {}));  | 
 | 268 | +                            })  | 
 | 269 | +                        .whenComplete((value, throwable) -> complete(result, value, throwable))));  | 
 | 270 | +    result.get(30, TimeUnit.SECONDS);  | 
 | 271 | + | 
 | 272 | +    assertTrace();  | 
 | 273 | +  }  | 
 | 274 | + | 
 | 275 | +  private void runWithVertx(Runnable runnable) {  | 
 | 276 | +    vertx.getOrCreateContext().runOnContext(event -> runnable.run());  | 
 | 277 | +  }  | 
 | 278 | + | 
 | 279 | +  private static void complete(  | 
 | 280 | +      CompletableFuture<Object> completableFuture, Object result, Throwable throwable) {  | 
 | 281 | +    if (throwable != null) {  | 
 | 282 | +      completableFuture.completeExceptionally(throwable);  | 
 | 283 | +    } else {  | 
 | 284 | +      completableFuture.complete(result);  | 
 | 285 | +    }  | 
 | 286 | +  }  | 
 | 287 | + | 
 | 288 | +  @SuppressWarnings("deprecation") // using deprecated semconv  | 
 | 289 | +  private void assertTrace() {  | 
 | 290 | +    testing.waitAndAssertTraces(  | 
 | 291 | +        trace ->  | 
 | 292 | +            trace.hasSpansSatisfyingExactly(  | 
 | 293 | +                span -> span.hasName("parent").hasKind(SpanKind.INTERNAL),  | 
 | 294 | +                span ->  | 
 | 295 | +                    span.hasName("SELECT tempdb.Value")  | 
 | 296 | +                        .hasKind(SpanKind.CLIENT)  | 
 | 297 | +                        .hasParent(trace.getSpan(0))  | 
 | 298 | +                        .hasAttributesSatisfyingExactly(  | 
 | 299 | +                            equalTo(maybeStable(DB_NAME), DB),  | 
 | 300 | +                            equalTo(DB_USER, emitStableDatabaseSemconv() ? null : USER_DB),  | 
 | 301 | +                            equalTo(  | 
 | 302 | +                                maybeStable(DB_STATEMENT),  | 
 | 303 | +                                "select v1_0.id,v1_0.name from Value v1_0 where v1_0.id=$1"),  | 
 | 304 | +                            equalTo(maybeStable(DB_OPERATION), "SELECT"),  | 
 | 305 | +                            equalTo(maybeStable(DB_SQL_TABLE), "Value"),  | 
 | 306 | +                            equalTo(SERVER_ADDRESS, host),  | 
 | 307 | +                            equalTo(SERVER_PORT, port)),  | 
 | 308 | +                span ->  | 
 | 309 | +                    span.hasName("callback")  | 
 | 310 | +                        .hasKind(SpanKind.INTERNAL)  | 
 | 311 | +                        .hasParent(trace.getSpan(0))));  | 
 | 312 | +  }  | 
 | 313 | +}  | 
0 commit comments