001/* 002 * Copyright 2026 Revetware LLC. 003 * 004 * Licensed under the Apache License, Version 2.0 (the "License"); 005 * you may not use this file except in compliance with the License. 006 * You may obtain a copy of the License at 007 * 008 * http://www.apache.org/licenses/LICENSE-2.0 009 * 010 * Unless required by applicable law or agreed to in writing, software 011 * distributed under the License is distributed on an "AS IS" BASIS, 012 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 013 * See the License for the specific language governing permissions and 014 * limitations under the License. 015 */ 016 017package com.pyranid.otel; 018 019import com.pyranid.DatabaseType; 020import com.pyranid.MetricsCollector; 021import com.pyranid.StatementContext; 022import com.pyranid.StatementLog; 023import com.pyranid.StatementResult; 024import com.pyranid.Transaction; 025import com.pyranid.TransactionIsolation; 026import com.pyranid.TransactionResult; 027import io.opentelemetry.api.GlobalOpenTelemetry; 028import io.opentelemetry.api.OpenTelemetry; 029import io.opentelemetry.api.common.AttributeKey; 030import io.opentelemetry.api.common.Attributes; 031import io.opentelemetry.api.common.AttributesBuilder; 032import io.opentelemetry.api.metrics.DoubleHistogram; 033import io.opentelemetry.api.metrics.LongCounter; 034import io.opentelemetry.api.metrics.LongHistogram; 035import io.opentelemetry.api.metrics.LongUpDownCounter; 036import io.opentelemetry.api.metrics.Meter; 037import io.opentelemetry.api.metrics.MeterBuilder; 038import org.jspecify.annotations.NonNull; 039import org.jspecify.annotations.Nullable; 040 041import javax.annotation.concurrent.NotThreadSafe; 042import javax.annotation.concurrent.ThreadSafe; 043import java.sql.SQLException; 044import java.time.Duration; 045import java.util.ArrayList; 046import java.util.Collections; 047import java.util.IdentityHashMap; 048import java.util.List; 049import java.util.Locale; 050import java.util.Map; 051import java.util.Optional; 052import java.util.Set; 053import java.util.UUID; 054import java.util.concurrent.ConcurrentHashMap; 055import java.util.concurrent.atomic.AtomicReferenceArray; 056 057import static java.util.Objects.requireNonNull; 058 059/** 060 * OpenTelemetry-backed {@link MetricsCollector} for Pyranid database metrics. 061 * <p> 062 * Standard database-client metrics use OpenTelemetry semantic-convention names directly. Pyranid-specific lifecycle 063 * events use {@code pyranid.*} metric and attribute names. 064 * <p> 065 * Notification metrics use bounded lifecycle attributes. Channel names, payloads, notification-session identifiers, 066 * and backend process identifiers are never exported. 067 * <p> 068 * This implementation is thread-safe and expects the supplied {@link Meter} to be backed by a non-blocking exporter. 069 * Pyranid core catches collector exceptions, but exporters that block can still stall the calling JDBC thread. 070 * 071 * @author <a href="https://www.revetkn.com">Mark Allen</a> 072 * @since 1.0.0 073 */ 074@ThreadSafe 075public final class OpenTelemetryMetricsCollector implements MetricsCollector { 076 @NonNull 077 private static final String DEFAULT_INSTRUMENTATION_NAME; 078 @NonNull 079 private static final String UNKNOWN_OPERATION; 080 @NonNull 081 private static final String OUTCOME_SUCCESS; 082 @NonNull 083 private static final String OUTCOME_FAILURE; 084 private static final int SQL_CLASSIFICATION_CACHE_CAPACITY; 085 private static final int SQL_CLASSIFICATION_CACHE_MASK; 086 private static final int SQL_CLASSIFICATION_CACHE_PROBES; 087 @NonNull 088 private static final List<Double> OPERATION_DURATION_BUCKET_BOUNDARIES; 089 @NonNull 090 private static final List<Long> RETURNED_ROWS_BUCKET_BOUNDARIES; 091 092 @NonNull 093 private static final AttributeKey<String> DB_SYSTEM_NAME_ATTRIBUTE_KEY; 094 @NonNull 095 private static final AttributeKey<String> DB_OPERATION_NAME_ATTRIBUTE_KEY; 096 @NonNull 097 private static final AttributeKey<String> DB_NAMESPACE_ATTRIBUTE_KEY; 098 @NonNull 099 private static final AttributeKey<String> DB_COLLECTION_NAME_ATTRIBUTE_KEY; 100 @NonNull 101 private static final AttributeKey<String> DB_RESPONSE_STATUS_CODE_ATTRIBUTE_KEY; 102 @NonNull 103 private static final AttributeKey<String> DB_CLIENT_CONNECTION_POOL_NAME_ATTRIBUTE_KEY; 104 @NonNull 105 private static final AttributeKey<String> ERROR_TYPE_ATTRIBUTE_KEY; 106 @NonNull 107 private static final AttributeKey<String> TRANSACTION_CLOSURE_OUTCOME_ATTRIBUTE_KEY; 108 @NonNull 109 private static final AttributeKey<String> TRANSACTION_COMMIT_OUTCOME_ATTRIBUTE_KEY; 110 @NonNull 111 private static final AttributeKey<String> TRANSACTION_ROLLBACK_OUTCOME_ATTRIBUTE_KEY; 112 @NonNull 113 private static final AttributeKey<String> TRANSACTION_BEGIN_PHASE_ATTRIBUTE_KEY; 114 @NonNull 115 private static final AttributeKey<String> TRANSACTION_RESULT_ATTRIBUTE_KEY; 116 @NonNull 117 private static final AttributeKey<String> TRANSACTION_ISOLATION_ATTRIBUTE_KEY; 118 @NonNull 119 private static final AttributeKey<String> SAVEPOINT_OPERATION_ATTRIBUTE_KEY; 120 @NonNull 121 private static final AttributeKey<String> STREAM_OUTCOME_ATTRIBUTE_KEY; 122 @NonNull 123 private static final AttributeKey<String> NOTIFICATION_SESSION_OPEN_OUTCOME_ATTRIBUTE_KEY; 124 @NonNull 125 private static final AttributeKey<String> NOTIFICATION_SESSION_OUTCOME_ATTRIBUTE_KEY; 126 127 static { 128 DEFAULT_INSTRUMENTATION_NAME = "com.pyranid.otel"; 129 UNKNOWN_OPERATION = "UNKNOWN"; 130 OUTCOME_SUCCESS = "success"; 131 OUTCOME_FAILURE = "failure"; 132 SQL_CLASSIFICATION_CACHE_CAPACITY = 1024; 133 SQL_CLASSIFICATION_CACHE_MASK = SQL_CLASSIFICATION_CACHE_CAPACITY - 1; 134 SQL_CLASSIFICATION_CACHE_PROBES = 4; 135 OPERATION_DURATION_BUCKET_BOUNDARIES = List.of(0.001D, 0.005D, 0.01D, 0.05D, 0.1D, 0.5D, 1D, 5D, 10D); 136 RETURNED_ROWS_BUCKET_BOUNDARIES = List.of(1L, 2L, 5L, 10L, 20L, 50L, 100L, 200L, 500L, 1_000L, 2_000L, 5_000L, 10_000L); 137 138 DB_SYSTEM_NAME_ATTRIBUTE_KEY = AttributeKey.stringKey("db.system.name"); 139 DB_OPERATION_NAME_ATTRIBUTE_KEY = AttributeKey.stringKey("db.operation.name"); 140 DB_NAMESPACE_ATTRIBUTE_KEY = AttributeKey.stringKey("db.namespace"); 141 DB_COLLECTION_NAME_ATTRIBUTE_KEY = AttributeKey.stringKey("db.collection.name"); 142 DB_RESPONSE_STATUS_CODE_ATTRIBUTE_KEY = AttributeKey.stringKey("db.response.status_code"); 143 DB_CLIENT_CONNECTION_POOL_NAME_ATTRIBUTE_KEY = AttributeKey.stringKey("db.client.connection.pool.name"); 144 ERROR_TYPE_ATTRIBUTE_KEY = AttributeKey.stringKey("error.type"); 145 TRANSACTION_CLOSURE_OUTCOME_ATTRIBUTE_KEY = AttributeKey.stringKey("pyranid.transaction.closure_outcome"); 146 TRANSACTION_COMMIT_OUTCOME_ATTRIBUTE_KEY = AttributeKey.stringKey("pyranid.transaction.commit_outcome"); 147 TRANSACTION_ROLLBACK_OUTCOME_ATTRIBUTE_KEY = AttributeKey.stringKey("pyranid.transaction.rollback_outcome"); 148 TRANSACTION_BEGIN_PHASE_ATTRIBUTE_KEY = AttributeKey.stringKey("pyranid.transaction.physical_begin_phase"); 149 TRANSACTION_RESULT_ATTRIBUTE_KEY = AttributeKey.stringKey("pyranid.transaction.result"); 150 TRANSACTION_ISOLATION_ATTRIBUTE_KEY = AttributeKey.stringKey("pyranid.transaction.isolation"); 151 SAVEPOINT_OPERATION_ATTRIBUTE_KEY = AttributeKey.stringKey("pyranid.savepoint.operation"); 152 STREAM_OUTCOME_ATTRIBUTE_KEY = AttributeKey.stringKey("pyranid.stream.outcome"); 153 NOTIFICATION_SESSION_OPEN_OUTCOME_ATTRIBUTE_KEY = AttributeKey.stringKey("pyranid.notification.session.open_outcome"); 154 NOTIFICATION_SESSION_OUTCOME_ATTRIBUTE_KEY = AttributeKey.stringKey("pyranid.notification.session.outcome"); 155 } 156 157 @NonNull 158 private final DoubleHistogram operationDurationHistogram; 159 @NonNull 160 private final LongHistogram returnedRowsHistogram; 161 @NonNull 162 private final DoubleHistogram connectionWaitTimeHistogram; 163 @NonNull 164 private final DoubleHistogram connectionUseTimeHistogram; 165 @NonNull 166 private final DoubleHistogram statementPreparationDurationHistogram; 167 @NonNull 168 private final DoubleHistogram statementExecutionDurationHistogram; 169 @NonNull 170 private final DoubleHistogram statementMappingDurationHistogram; 171 @NonNull 172 private final LongCounter statementErrorsCounter; 173 @NonNull 174 private final LongHistogram statementBatchSizeHistogram; 175 @NonNull 176 private final LongHistogram statementRowsAffectedHistogram; 177 @NonNull 178 private final DoubleHistogram transactionClosureDurationHistogram; 179 @NonNull 180 private final DoubleHistogram transactionCommitDurationHistogram; 181 @NonNull 182 private final DoubleHistogram transactionRollbackDurationHistogram; 183 @NonNull 184 private final LongCounter transactionPhysicalBeginFailuresCounter; 185 @NonNull 186 private final LongCounter transactionCountCounter; 187 @NonNull 188 private final LongUpDownCounter activeTransactionsCounter; 189 @NonNull 190 private final Set<Transaction> activeTransactions; 191 @NonNull 192 private final LongCounter savepointOperationsCounter; 193 @NonNull 194 private final DoubleHistogram streamDurationHistogram; 195 @NonNull 196 private final LongHistogram streamRowsConsumedHistogram; 197 @NonNull 198 private final LongCounter postTransactionOperationsCounter; 199 @NonNull 200 private final DoubleHistogram postTransactionDurationHistogram; 201 @NonNull 202 private final DoubleHistogram notificationSessionOpenDurationHistogram; 203 @NonNull 204 private final DoubleHistogram notificationSessionDurationHistogram; 205 @NonNull 206 private final LongUpDownCounter activeNotificationSessionsCounter; 207 @NonNull 208 private final Map<UUID, Attributes> activeNotificationSessions; 209 @NonNull 210 private final LongHistogram notificationBatchSizeHistogram; 211 @NonNull 212 private final LongCounter notificationConnectionLossesCounter; 213 @Nullable 214 private final String poolName; 215 @Nullable 216 private final String namespace; 217 @NonNull 218 private final Boolean recordCollectionName; 219 @NonNull 220 private final Boolean recordFullSqlState; 221 @NonNull 222 private final AtomicReferenceArray<SqlClassificationCacheEntry> sqlClassificationCache; 223 224 /** 225 * Acquires a builder for {@link OpenTelemetryMetricsCollector} instances, using {@link GlobalOpenTelemetry} 226 * by default. 227 * 228 * @return the builder 229 */ 230 @NonNull 231 public static Builder builder() { 232 return new Builder(); 233 } 234 235 /** 236 * Acquires a builder seeded with a required {@link Meter}. 237 * 238 * @param meter the meter used to build instruments 239 * @return the builder 240 */ 241 @NonNull 242 public static Builder withMeter(@NonNull Meter meter) { 243 requireNonNull(meter); 244 return builder().meter(meter); 245 } 246 247 /** 248 * Acquires a builder seeded with a required {@link OpenTelemetry} instance. 249 * 250 * @param openTelemetry the OpenTelemetry instance used to build a meter 251 * @return the builder 252 */ 253 @NonNull 254 public static Builder withOpenTelemetry(@NonNull OpenTelemetry openTelemetry) { 255 requireNonNull(openTelemetry); 256 return builder().openTelemetry(openTelemetry); 257 } 258 259 /** 260 * Creates an instance from a required {@link Meter} without additional customization. 261 * 262 * @param meter the meter used to build instruments 263 * @return an {@link OpenTelemetryMetricsCollector} instance 264 */ 265 @NonNull 266 public static OpenTelemetryMetricsCollector fromMeter(@NonNull Meter meter) { 267 return withMeter(meter).build(); 268 } 269 270 /** 271 * Creates an instance from a required {@link OpenTelemetry} without additional customization. 272 * 273 * @param openTelemetry the OpenTelemetry instance used to build a meter 274 * @return an {@link OpenTelemetryMetricsCollector} instance 275 */ 276 @NonNull 277 public static OpenTelemetryMetricsCollector fromOpenTelemetry(@NonNull OpenTelemetry openTelemetry) { 278 return withOpenTelemetry(openTelemetry).build(); 279 } 280 281 private OpenTelemetryMetricsCollector(@NonNull Builder builder) { 282 requireNonNull(builder); 283 Meter meter = requireNonNull(builder.resolveMeter()); 284 this.poolName = builder.poolName; 285 this.namespace = builder.namespace; 286 this.recordCollectionName = builder.recordCollectionName; 287 this.recordFullSqlState = builder.recordFullSqlState; 288 this.sqlClassificationCache = new AtomicReferenceArray<>(SQL_CLASSIFICATION_CACHE_CAPACITY); 289 290 this.operationDurationHistogram = meter.histogramBuilder("db.client.operation.duration") 291 .setDescription("Duration of database client operations.") 292 .setUnit("s") 293 .setExplicitBucketBoundariesAdvice(OPERATION_DURATION_BUCKET_BOUNDARIES) 294 .build(); 295 this.returnedRowsHistogram = meter.histogramBuilder("db.client.response.returned_rows") 296 .ofLongs() 297 .setDescription("Number of rows returned by a database response.") 298 .setUnit("{row}") 299 .setExplicitBucketBoundariesAdvice(RETURNED_ROWS_BUCKET_BOUNDARIES) 300 .build(); 301 this.connectionWaitTimeHistogram = meter.histogramBuilder("db.client.connection.wait_time") 302 .setDescription("Time spent waiting for a database connection.") 303 .setUnit("s") 304 .build(); 305 this.connectionUseTimeHistogram = meter.histogramBuilder("db.client.connection.use_time") 306 .setDescription("Time a database connection was held by Pyranid.") 307 .setUnit("s") 308 .build(); 309 this.statementPreparationDurationHistogram = meter.histogramBuilder("pyranid.statement.preparation.duration") 310 .setDescription("Duration spent preparing and binding a JDBC statement.") 311 .setUnit("s") 312 .build(); 313 this.statementExecutionDurationHistogram = meter.histogramBuilder("pyranid.statement.execution.duration") 314 .setDescription("Duration spent executing a JDBC statement.") 315 .setUnit("s") 316 .build(); 317 this.statementMappingDurationHistogram = meter.histogramBuilder("pyranid.statement.mapping.duration") 318 .setDescription("Duration spent mapping JDBC result-set rows.") 319 .setUnit("s") 320 .build(); 321 this.statementErrorsCounter = meter.counterBuilder("pyranid.statement.errors") 322 .setDescription("Total number of failed Pyranid statement operations.") 323 .setUnit("{statement}") 324 .build(); 325 this.statementBatchSizeHistogram = meter.histogramBuilder("pyranid.statement.batch.size") 326 .ofLongs() 327 .setDescription("Number of statement parameter groups submitted in a batch.") 328 .setUnit("{statement}") 329 .build(); 330 this.statementRowsAffectedHistogram = meter.histogramBuilder("pyranid.statement.rows_affected") 331 .ofLongs() 332 .setDescription("Rows affected by DML/update statements.") 333 .setUnit("{row}") 334 .build(); 335 this.transactionClosureDurationHistogram = meter.histogramBuilder("pyranid.transaction.closure.duration") 336 .setDescription("Logical Pyranid transaction closure duration.") 337 .setUnit("s") 338 .build(); 339 this.transactionCommitDurationHistogram = meter.histogramBuilder("pyranid.transaction.commit.duration") 340 .setDescription("JDBC transaction commit operation duration.") 341 .setUnit("s") 342 .build(); 343 this.transactionRollbackDurationHistogram = meter.histogramBuilder("pyranid.transaction.rollback.duration") 344 .setDescription("JDBC transaction rollback operation duration.") 345 .setUnit("s") 346 .build(); 347 this.transactionPhysicalBeginFailuresCounter = meter.counterBuilder("pyranid.transaction.physical.begin_failures") 348 .setDescription("Total number of physical transaction begin failures.") 349 .setUnit("{transaction}") 350 .build(); 351 this.transactionCountCounter = meter.counterBuilder("pyranid.transaction.count") 352 .setDescription("Total number of logical transaction closures.") 353 .setUnit("{transaction}") 354 .build(); 355 this.activeTransactionsCounter = meter.upDownCounterBuilder("pyranid.transaction.active") 356 .setDescription("Number of active physical JDBC transactions.") 357 .setUnit("{transaction}") 358 .build(); 359 this.activeTransactions = ConcurrentHashMap.newKeySet(); 360 this.savepointOperationsCounter = meter.counterBuilder("pyranid.savepoint.operations") 361 .setDescription("Total number of savepoint operations.") 362 .setUnit("{operation}") 363 .build(); 364 this.streamDurationHistogram = meter.histogramBuilder("pyranid.fetchstream.duration") 365 .setDescription("Duration of Pyranid fetchStream consumption.") 366 .setUnit("s") 367 .build(); 368 this.streamRowsConsumedHistogram = meter.histogramBuilder("pyranid.fetchstream.rows_consumed") 369 .ofLongs() 370 .setDescription("Rows consumed through Pyranid fetchStream.") 371 .setUnit("{row}") 372 .build(); 373 this.postTransactionOperationsCounter = meter.counterBuilder("pyranid.post_transaction.operations") 374 .setDescription("Total number of post-transaction operations.") 375 .setUnit("{operation}") 376 .build(); 377 this.postTransactionDurationHistogram = meter.histogramBuilder("pyranid.post_transaction.duration") 378 .setDescription("Duration of post-transaction operations.") 379 .setUnit("s") 380 .build(); 381 this.notificationSessionOpenDurationHistogram = meter.histogramBuilder("pyranid.notification.session.open.duration") 382 .setDescription("Duration of notification-session opening attempts.") 383 .setUnit("s") 384 .build(); 385 this.notificationSessionDurationHistogram = meter.histogramBuilder("pyranid.notification.session.duration") 386 .setDescription("Duration of opened notification sessions, including cleanup.") 387 .setUnit("s") 388 .build(); 389 this.activeNotificationSessionsCounter = meter.upDownCounterBuilder("pyranid.notification.session.active") 390 .setDescription("Number of active notification sessions.") 391 .setUnit("{session}") 392 .build(); 393 this.activeNotificationSessions = new ConcurrentHashMap<>(); 394 this.notificationBatchSizeHistogram = meter.histogramBuilder("pyranid.notification.batch.size") 395 .ofLongs() 396 .setDescription("Number of notifications delivered to application code in a batch.") 397 .setUnit("{notification}") 398 .build(); 399 this.notificationConnectionLossesCounter = meter.counterBuilder("pyranid.notification.connection.losses") 400 .setDescription("Total number of terminal notification-connection losses.") 401 .setUnit("{connection}") 402 .build(); 403 } 404 405 @Override 406 public void didAcquireStatementConnection(@NonNull StatementContext<?> ctx, 407 @NonNull Duration acquisitionDuration) { 408 if (this.poolName == null) 409 return; 410 411 requireNonNull(ctx); 412 recordConnectionWaitTime(ctx.getDatabaseType(), acquisitionDuration, null); 413 } 414 415 @Override 416 public void didFailToAcquireStatementConnection(@NonNull StatementContext<?> ctx, 417 @NonNull DatabaseType databaseType, 418 @NonNull Duration acquisitionDuration, 419 @NonNull Throwable throwable) { 420 if (this.poolName == null) 421 return; 422 423 requireNonNull(ctx); 424 requireNonNull(databaseType); 425 recordConnectionWaitTime(databaseType, acquisitionDuration, requireNonNull(throwable)); 426 } 427 428 @Override 429 public void didAcquireTransactionConnection(@NonNull Transaction transaction, 430 @NonNull DatabaseType databaseType, 431 @NonNull Duration acquisitionDuration) { 432 recordConnectionWaitTime(databaseType, acquisitionDuration, null); 433 } 434 435 @Override 436 public void didFailToAcquireTransactionConnection(@NonNull Transaction transaction, 437 @NonNull DatabaseType databaseType, 438 @NonNull Duration acquisitionDuration, 439 @NonNull Throwable throwable) { 440 recordConnectionWaitTime(databaseType, acquisitionDuration, requireNonNull(throwable)); 441 } 442 443 @Override 444 public void didReleaseStatementConnection(@NonNull StatementContext<?> ctx, 445 @NonNull Duration heldDuration) { 446 if (this.poolName == null) 447 return; 448 449 requireNonNull(ctx); 450 recordConnectionUseTime(ctx.getDatabaseType(), heldDuration, null); 451 } 452 453 @Override 454 public void didFailToReleaseStatementConnection(@NonNull StatementContext<?> ctx, 455 @NonNull Duration heldDuration, 456 @NonNull Throwable throwable) { 457 if (this.poolName == null) 458 return; 459 460 requireNonNull(ctx); 461 recordConnectionUseTime(ctx.getDatabaseType(), heldDuration, requireNonNull(throwable)); 462 } 463 464 @Override 465 public void didReleaseTransactionConnection(@NonNull Transaction transaction, 466 @NonNull DatabaseType databaseType, 467 @NonNull Duration heldDuration) { 468 recordConnectionUseTime(databaseType, heldDuration, null); 469 } 470 471 @Override 472 public void didFailToReleaseTransactionConnection(@NonNull Transaction transaction, 473 @NonNull DatabaseType databaseType, 474 @NonNull Duration heldDuration, 475 @NonNull Throwable throwable) { 476 recordConnectionUseTime(databaseType, heldDuration, requireNonNull(throwable)); 477 } 478 479 @Override 480 public void didExitTransactionClosure(@NonNull Transaction transaction, 481 @NonNull TransactionClosureOutcome outcome, 482 @NonNull DatabaseType databaseType, 483 @NonNull Duration logicalDuration, 484 @Nullable Throwable thrown) { 485 requireNonNull(transaction); 486 requireNonNull(databaseType); 487 488 try { 489 requireNonNull(outcome); 490 Attributes attributes = transactionClosureAttributes(databaseType, transaction.getTransactionIsolation(), outcome, thrown); 491 this.transactionClosureDurationHistogram.record(seconds(logicalDuration), attributes); 492 this.transactionCountCounter.add(1, attributes); 493 } finally { 494 // This is an idempotent fallback when a physical commit/rollback callback is absent or cannot finish. 495 markPhysicalTransactionInactive(transaction, databaseType); 496 } 497 } 498 499 @Override 500 public void didBeginPhysicalTransaction(@NonNull Transaction transaction, 501 @NonNull TransactionIsolation isolation, 502 @NonNull DatabaseType databaseType) { 503 markPhysicalTransactionActive(transaction, databaseType); 504 } 505 506 @Override 507 public void didFailToBeginPhysicalTransaction(@NonNull Transaction transaction, 508 @NonNull TransactionIsolation isolation, 509 @NonNull PhysicalTransactionBeginFailurePhase phase, 510 @NonNull DatabaseType databaseType, 511 @NonNull Throwable throwable) { 512 Attributes attributes = Attributes.builder() 513 .putAll(databaseAttributes(databaseType)) 514 .put(TRANSACTION_ISOLATION_ATTRIBUTE_KEY, enumValue(isolation)) 515 .put(TRANSACTION_BEGIN_PHASE_ATTRIBUTE_KEY, enumValue(phase)) 516 .put(ERROR_TYPE_ATTRIBUTE_KEY, errorType(throwable)) 517 .build(); 518 this.transactionPhysicalBeginFailuresCounter.add(1, attributes); 519 } 520 521 @Override 522 public void didCommitPhysicalTransaction(@NonNull Transaction transaction, 523 @NonNull DatabaseType databaseType, 524 @NonNull Duration physicalDuration) { 525 this.transactionCommitDurationHistogram.record(seconds(physicalDuration), 526 transactionOperationAttributes(databaseType, transaction.getTransactionIsolation(), TRANSACTION_COMMIT_OUTCOME_ATTRIBUTE_KEY, OUTCOME_SUCCESS, null)); 527 markPhysicalTransactionInactive(transaction, databaseType); 528 } 529 530 @Override 531 public void didFailToCommitPhysicalTransaction(@NonNull Transaction transaction, 532 @NonNull DatabaseType databaseType, 533 @NonNull Duration physicalDuration, 534 @NonNull Throwable throwable) { 535 this.transactionCommitDurationHistogram.record(seconds(physicalDuration), 536 transactionOperationAttributes(databaseType, transaction.getTransactionIsolation(), TRANSACTION_COMMIT_OUTCOME_ATTRIBUTE_KEY, OUTCOME_FAILURE, throwable)); 537 // No active-transaction decrement here: Pyranid attempts rollback after commit failure, and the following 538 // rollback callback owns the single terminal decrement. Decrementing here would double-count. 539 } 540 541 @Override 542 public void didRollbackPhysicalTransaction(@NonNull Transaction transaction, 543 @NonNull DatabaseType databaseType, 544 @NonNull Duration physicalDuration) { 545 this.transactionRollbackDurationHistogram.record(seconds(physicalDuration), 546 transactionOperationAttributes(databaseType, transaction.getTransactionIsolation(), TRANSACTION_ROLLBACK_OUTCOME_ATTRIBUTE_KEY, OUTCOME_SUCCESS, null)); 547 markPhysicalTransactionInactive(transaction, databaseType); 548 } 549 550 @Override 551 public void didFailToRollbackPhysicalTransaction(@NonNull Transaction transaction, 552 @NonNull DatabaseType databaseType, 553 @NonNull Duration physicalDuration, 554 @NonNull Throwable throwable) { 555 this.transactionRollbackDurationHistogram.record(seconds(physicalDuration), 556 transactionOperationAttributes(databaseType, transaction.getTransactionIsolation(), TRANSACTION_ROLLBACK_OUTCOME_ATTRIBUTE_KEY, OUTCOME_FAILURE, throwable)); 557 markPhysicalTransactionInactive(transaction, databaseType); 558 } 559 560 @Override 561 public void didCreateSavepoint(@NonNull Transaction transaction, 562 @NonNull DatabaseType databaseType) { 563 recordSavepointOperation(databaseType, "created"); 564 } 565 566 @Override 567 public void didRollbackToSavepoint(@NonNull Transaction transaction, 568 @NonNull DatabaseType databaseType) { 569 recordSavepointOperation(databaseType, "rolled_back"); 570 } 571 572 @Override 573 public void didReleaseSavepoint(@NonNull Transaction transaction, 574 @NonNull DatabaseType databaseType) { 575 recordSavepointOperation(databaseType, "released"); 576 } 577 578 @Override 579 public void didExecuteStatement(@NonNull StatementContext<?> ctx, 580 @NonNull StatementLog<?> statementLog, 581 @NonNull StatementResult result) { 582 requireNonNull(ctx); 583 requireNonNull(statementLog); 584 requireNonNull(result); 585 586 Integer batchSize = statementLog.getBatchSize().orElse(null); 587 Attributes attributes = statementAttributes(ctx, null, null, batchSize); 588 this.operationDurationHistogram.record(seconds(statementLog.getTotalDuration()), attributes); 589 recordStatementComponentDurations(statementLog, attributes); 590 591 if (batchSize != null) 592 this.statementBatchSizeHistogram.record(batchSize.longValue(), attributes); 593 594 Long rowsReturned = result.rowsReturned(); 595 if (rowsReturned != null) 596 this.returnedRowsHistogram.record(rowsReturned, attributes); 597 598 Long rowsAffected = result.rowsAffected(); 599 if (rowsAffected != null) 600 this.statementRowsAffectedHistogram.record(rowsAffected, attributes); 601 } 602 603 @Override 604 public void didFailToExecuteStatement(@NonNull StatementContext<?> ctx, 605 @NonNull StatementLog<?> statementLog, 606 @NonNull DatabaseType databaseType, 607 @NonNull Throwable throwable) { 608 requireNonNull(ctx); 609 requireNonNull(statementLog); 610 requireNonNull(databaseType); 611 requireNonNull(throwable); 612 613 Integer batchSize = statementLog.getBatchSize().orElse(null); 614 Attributes attributes = statementAttributes(ctx, throwable, databaseType, batchSize); 615 this.operationDurationHistogram.record(seconds(statementLog.getTotalDuration()), attributes); 616 recordStatementComponentDurations(statementLog, attributes); 617 if (batchSize != null) 618 this.statementBatchSizeHistogram.record(batchSize.longValue(), attributes); 619 this.statementErrorsCounter.add(1, attributes); 620 } 621 622 @Override 623 public void didFailToOpenStream(@NonNull StatementContext<?> ctx, 624 @NonNull DatabaseType databaseType, 625 @NonNull Duration openDuration, 626 @NonNull Throwable throwable) { 627 requireNonNull(ctx); 628 requireNonNull(databaseType); 629 requireNonNull(throwable); 630 631 Attributes attributes = streamAttributes(databaseType, StreamTerminalOutcome.OPEN_FAILURE, throwable); 632 this.streamDurationHistogram.record(seconds(openDuration), attributes); 633 } 634 635 @Override 636 public void didCloseStream(@NonNull StatementContext<?> ctx, 637 @NonNull StreamTerminalOutcome outcome, 638 @NonNull Long rowsConsumed, 639 @NonNull Duration streamDuration, 640 @Nullable Throwable throwable) { 641 requireNonNull(ctx); 642 requireNonNull(outcome); 643 requireNonNull(rowsConsumed); 644 requireNonNull(streamDuration); 645 646 Attributes attributes = streamAttributes(ctx.getDatabaseType(), outcome, throwable); 647 this.streamDurationHistogram.record(seconds(streamDuration), attributes); 648 this.streamRowsConsumedHistogram.record(rowsConsumed, attributes); 649 } 650 651 @Override 652 public void willOpenNotificationSession(@NonNull DatabaseType databaseType, 653 @NonNull UUID notificationSessionId) { 654 // The matching callbacks supply the authoritative duration and outcome; there is no metric to record yet. 655 requireNonNull(databaseType); 656 requireNonNull(notificationSessionId); 657 } 658 659 @Override 660 public void didOpenNotificationSession(@NonNull DatabaseType databaseType, 661 @NonNull UUID notificationSessionId, 662 @NonNull Duration openDuration) { 663 requireNonNull(openDuration); 664 markNotificationSessionActive(databaseType, notificationSessionId); 665 this.notificationSessionOpenDurationHistogram.record(seconds(openDuration), 666 notificationAttributes(databaseType, NOTIFICATION_SESSION_OPEN_OUTCOME_ATTRIBUTE_KEY, OUTCOME_SUCCESS, null)); 667 } 668 669 @Override 670 public void didFailToOpenNotificationSession(@NonNull DatabaseType databaseType, 671 @NonNull UUID notificationSessionId, 672 @NonNull Duration openDuration, 673 @NonNull Throwable throwable) { 674 requireNonNull(notificationSessionId); 675 requireNonNull(openDuration); 676 requireNonNull(throwable); 677 this.notificationSessionOpenDurationHistogram.record(seconds(openDuration), 678 notificationAttributes(databaseType, NOTIFICATION_SESSION_OPEN_OUTCOME_ATTRIBUTE_KEY, OUTCOME_FAILURE, throwable)); 679 } 680 681 @Override 682 public void didDeliverNotificationBatch(@NonNull DatabaseType databaseType, 683 @NonNull UUID notificationSessionId, 684 @NonNull Long notificationCount) { 685 requireNonNull(notificationSessionId); 686 this.notificationBatchSizeHistogram.record(requireNonNull(notificationCount), notificationAttributes(databaseType)); 687 } 688 689 @Override 690 public void didLoseNotificationConnection(@NonNull DatabaseType databaseType, 691 @NonNull UUID notificationSessionId, 692 @NonNull Throwable throwable) { 693 requireNonNull(notificationSessionId); 694 this.notificationConnectionLossesCounter.add(1, 695 notificationAttributes(databaseType, null, null, requireNonNull(throwable))); 696 } 697 698 @Override 699 public void didCloseNotificationSession(@NonNull DatabaseType databaseType, 700 @NonNull UUID notificationSessionId, 701 @NonNull NotificationSessionOutcome outcome, 702 @NonNull Duration sessionDuration, 703 @Nullable Throwable throwable) { 704 requireNonNull(outcome); 705 requireNonNull(sessionDuration); 706 707 try { 708 this.notificationSessionDurationHistogram.record(seconds(sessionDuration), 709 notificationAttributes(databaseType, NOTIFICATION_SESSION_OUTCOME_ATTRIBUTE_KEY, enumValue(outcome), throwable)); 710 } finally { 711 markNotificationSessionInactive(notificationSessionId); 712 } 713 } 714 715 @Override 716 public void didRunPostTransactionOperation(@NonNull Transaction transaction, 717 @NonNull TransactionResult result, 718 @NonNull DatabaseType databaseType, 719 @NonNull Duration duration, 720 @Nullable Throwable throwable) { 721 AttributesBuilder builder = Attributes.builder() 722 .putAll(databaseAttributes(databaseType)) 723 .put(TRANSACTION_RESULT_ATTRIBUTE_KEY, enumValue(result)); 724 725 if (throwable != null) 726 builder.put(ERROR_TYPE_ATTRIBUTE_KEY, errorType(throwable)); 727 728 Attributes attributes = builder.build(); 729 this.postTransactionOperationsCounter.add(1, attributes); 730 this.postTransactionDurationHistogram.record(seconds(duration), attributes); 731 } 732 733 @Override 734 @NonNull 735 public Optional<Snapshot> snapshot() { 736 return Optional.empty(); 737 } 738 739 @NonNull 740 private Attributes statementAttributes(@NonNull StatementContext<?> ctx, 741 @Nullable Throwable throwable, 742 @Nullable DatabaseType databaseTypeOverride, 743 @Nullable Integer batchSize) { 744 requireNonNull(ctx); 745 SqlClassification sqlClassification = sqlClassificationFor(ctx.getStatement().getSql()); 746 747 AttributesBuilder builder = Attributes.builder() 748 .putAll(databaseAttributes(databaseTypeOverride == null ? ctx.getDatabaseType() : databaseTypeOverride)) 749 .put(DB_OPERATION_NAME_ATTRIBUTE_KEY, operationName(sqlClassification.operation(), batchSize != null)); 750 751 if (this.namespace != null) 752 builder.put(DB_NAMESPACE_ATTRIBUTE_KEY, this.namespace); 753 754 if (this.recordCollectionName) { 755 String collectionName = sqlClassification.collectionName(); 756 if (collectionName != null) 757 builder.put(DB_COLLECTION_NAME_ATTRIBUTE_KEY, collectionName); 758 } 759 760 String statusCode = dbResponseStatusCode(throwable); 761 if (statusCode != null) 762 builder.put(DB_RESPONSE_STATUS_CODE_ATTRIBUTE_KEY, statusCode); 763 764 if (throwable != null) 765 builder.put(ERROR_TYPE_ATTRIBUTE_KEY, errorType(throwable)); 766 767 return builder.build(); 768 } 769 770 @NonNull 771 private Attributes databaseAttributes(@NonNull DatabaseType databaseType) { 772 requireNonNull(databaseType); 773 return Attributes.of(DB_SYSTEM_NAME_ATTRIBUTE_KEY, dbSystemName(databaseType)); 774 } 775 776 @NonNull 777 private Attributes notificationAttributes(@NonNull DatabaseType databaseType) { 778 return notificationAttributes(databaseType, null, null, null); 779 } 780 781 @NonNull 782 private Attributes notificationAttributes(@NonNull DatabaseType databaseType, 783 @Nullable AttributeKey<String> outcomeAttributeKey, 784 @Nullable String outcome, 785 @Nullable Throwable throwable) { 786 AttributesBuilder builder = Attributes.builder().putAll(databaseAttributes(databaseType)); 787 788 if (this.namespace != null) 789 builder.put(DB_NAMESPACE_ATTRIBUTE_KEY, this.namespace); 790 791 if (this.poolName != null) 792 builder.put(DB_CLIENT_CONNECTION_POOL_NAME_ATTRIBUTE_KEY, this.poolName); 793 794 if (outcomeAttributeKey != null) 795 builder.put(outcomeAttributeKey, requireNonNull(outcome)); 796 797 if (throwable != null) 798 builder.put(ERROR_TYPE_ATTRIBUTE_KEY, errorType(throwable)); 799 800 return builder.build(); 801 } 802 803 private void markNotificationSessionActive(@NonNull DatabaseType databaseType, 804 @NonNull UUID notificationSessionId) { 805 requireNonNull(notificationSessionId); 806 Attributes attributes = notificationAttributes(databaseType); 807 808 if (this.activeNotificationSessions.putIfAbsent(notificationSessionId, attributes) != null) 809 return; 810 811 try { 812 this.activeNotificationSessionsCounter.add(1, attributes); 813 } catch (RuntimeException | Error throwable) { 814 this.activeNotificationSessions.remove(notificationSessionId, attributes); 815 throw throwable; 816 } 817 } 818 819 private void markNotificationSessionInactive(@NonNull UUID notificationSessionId) { 820 requireNonNull(notificationSessionId); 821 Attributes attributes = this.activeNotificationSessions.remove(notificationSessionId); 822 823 if (attributes == null) 824 return; 825 826 try { 827 this.activeNotificationSessionsCounter.add(-1, attributes); 828 } catch (RuntimeException | Error throwable) { 829 this.activeNotificationSessions.putIfAbsent(notificationSessionId, attributes); 830 throw throwable; 831 } 832 } 833 834 private void markPhysicalTransactionActive(@NonNull Transaction transaction, 835 @NonNull DatabaseType databaseType) { 836 requireNonNull(transaction); 837 requireNonNull(databaseType); 838 839 if (!this.activeTransactions.add(transaction)) 840 return; 841 842 try { 843 this.activeTransactionsCounter.add(1, databaseAttributes(databaseType)); 844 } catch (RuntimeException | Error throwable) { 845 this.activeTransactions.remove(transaction); 846 throw throwable; 847 } 848 } 849 850 private void markPhysicalTransactionInactive(@NonNull Transaction transaction, 851 @NonNull DatabaseType databaseType) { 852 requireNonNull(transaction); 853 requireNonNull(databaseType); 854 855 if (!this.activeTransactions.remove(transaction)) 856 return; 857 858 try { 859 this.activeTransactionsCounter.add(-1, databaseAttributes(databaseType)); 860 } catch (RuntimeException | Error throwable) { 861 this.activeTransactions.add(transaction); 862 throw throwable; 863 } 864 } 865 866 @NonNull 867 private Attributes transactionClosureAttributes(@NonNull DatabaseType databaseType, 868 @NonNull TransactionIsolation isolation, 869 @NonNull TransactionClosureOutcome outcome, 870 @Nullable Throwable throwable) { 871 AttributesBuilder builder = Attributes.builder() 872 .putAll(databaseAttributes(databaseType)) 873 .put(TRANSACTION_ISOLATION_ATTRIBUTE_KEY, enumValue(isolation)) 874 .put(TRANSACTION_CLOSURE_OUTCOME_ATTRIBUTE_KEY, enumValue(outcome)); 875 876 if (throwable != null) 877 builder.put(ERROR_TYPE_ATTRIBUTE_KEY, errorType(throwable)); 878 879 return builder.build(); 880 } 881 882 @NonNull 883 private Attributes transactionOperationAttributes(@NonNull DatabaseType databaseType, 884 @NonNull TransactionIsolation isolation, 885 @NonNull AttributeKey<String> outcomeAttributeKey, 886 @NonNull String outcome, 887 @Nullable Throwable throwable) { 888 AttributesBuilder builder = Attributes.builder() 889 .putAll(databaseAttributes(databaseType)) 890 .put(TRANSACTION_ISOLATION_ATTRIBUTE_KEY, enumValue(isolation)) 891 .put(outcomeAttributeKey, outcome); 892 893 if (throwable != null) 894 builder.put(ERROR_TYPE_ATTRIBUTE_KEY, errorType(throwable)); 895 896 return builder.build(); 897 } 898 899 @NonNull 900 private Attributes streamAttributes(@NonNull DatabaseType databaseType, 901 @NonNull StreamTerminalOutcome outcome, 902 @Nullable Throwable throwable) { 903 AttributesBuilder builder = Attributes.builder() 904 .putAll(databaseAttributes(databaseType)) 905 .put(STREAM_OUTCOME_ATTRIBUTE_KEY, enumValue(outcome)); 906 907 if (throwable != null) 908 builder.put(ERROR_TYPE_ATTRIBUTE_KEY, errorType(throwable)); 909 910 return builder.build(); 911 } 912 913 private void recordStatementComponentDurations(@NonNull StatementLog<?> statementLog, 914 @NonNull Attributes attributes) { 915 requireNonNull(statementLog); 916 requireNonNull(attributes); 917 statementLog.getPreparationDuration().ifPresent(duration -> this.statementPreparationDurationHistogram.record(seconds(duration), attributes)); 918 statementLog.getExecutionDuration().ifPresent(duration -> this.statementExecutionDurationHistogram.record(seconds(duration), attributes)); 919 statementLog.getResultSetMappingDuration().ifPresent(duration -> this.statementMappingDurationHistogram.record(seconds(duration), attributes)); 920 } 921 922 private void recordConnectionWaitTime(@NonNull DatabaseType databaseType, 923 @NonNull Duration duration, 924 @Nullable Throwable throwable) { 925 requireNonNull(databaseType); 926 requireNonNull(duration); 927 928 if (this.poolName == null) 929 return; 930 931 this.connectionWaitTimeHistogram.record(seconds(duration), connectionAttributes(databaseType, throwable)); 932 } 933 934 private void recordConnectionUseTime(@NonNull DatabaseType databaseType, 935 @NonNull Duration duration, 936 @Nullable Throwable throwable) { 937 requireNonNull(databaseType); 938 requireNonNull(duration); 939 940 if (this.poolName == null) 941 return; 942 943 this.connectionUseTimeHistogram.record(seconds(duration), connectionAttributes(databaseType, throwable)); 944 } 945 946 @NonNull 947 private Attributes connectionAttributes(@NonNull DatabaseType databaseType, 948 @Nullable Throwable throwable) { 949 AttributesBuilder builder = Attributes.builder() 950 .put(DB_CLIENT_CONNECTION_POOL_NAME_ATTRIBUTE_KEY, requireNonNull(this.poolName)) 951 .putAll(databaseAttributes(databaseType)); 952 953 if (throwable != null) 954 builder.put(ERROR_TYPE_ATTRIBUTE_KEY, errorType(throwable)); 955 956 return builder.build(); 957 } 958 959 private void recordSavepointOperation(@NonNull DatabaseType databaseType, 960 @NonNull String operation) { 961 this.savepointOperationsCounter.add(1, Attributes.builder() 962 .putAll(databaseAttributes(databaseType)) 963 .put(SAVEPOINT_OPERATION_ATTRIBUTE_KEY, operation) 964 .build()); 965 } 966 967 @NonNull 968 static String dbSystemName(@NonNull DatabaseType databaseType) { 969 requireNonNull(databaseType); 970 // Maps to OpenTelemetry's db.system.name registry values. A default arm keeps this forward-compatible: a future 971 // DatabaseType added to the core falls back to "other_sql" instead of throwing at runtime (the previous switch had 972 // no default, so values added after this module was compiled would fail). 973 return switch (databaseType) { 974 case POSTGRESQL -> "postgresql"; 975 case MYSQL -> "mysql"; 976 case MARIA_DB -> "mariadb"; 977 case SQLITE -> "sqlite"; 978 case SQL_SERVER -> "microsoft.sql_server"; 979 case ORACLE -> "oracle.db"; 980 default -> "other_sql"; // GENERIC and any future DatabaseType 981 }; 982 } 983 984 @NonNull 985 static String operationNameForSql(@NonNull String sql, 986 boolean batch) { 987 String operation = analyzeSql(requireNonNull(sql)).operation(); 988 return operationName(operation, batch); 989 } 990 991 @NonNull 992 private static String operationName(@NonNull String operation, 993 boolean batch) { 994 requireNonNull(operation); 995 996 if (!batch) 997 return operation; 998 999 return UNKNOWN_OPERATION.equals(operation) ? "BATCH" : "BATCH " + operation; 1000 } 1001 1002 @Nullable 1003 static String collectionNameForSql(@NonNull String sql) { 1004 return collectionName(analyzeSql(requireNonNull(sql))); 1005 } 1006 1007 @NonNull 1008 private SqlClassification sqlClassificationFor(@NonNull String sql) { 1009 requireNonNull(sql); 1010 int firstIndex = spreadHash(sql.hashCode()) & SQL_CLASSIFICATION_CACHE_MASK; 1011 int emptyIndex = -1; 1012 1013 // Adjacent probes let colliding hot statements coexist while keeping the cache bounded and lock-free. 1014 for (int offset = 0; offset < SQL_CLASSIFICATION_CACHE_PROBES; offset++) { 1015 int index = (firstIndex + offset) & SQL_CLASSIFICATION_CACHE_MASK; 1016 SqlClassificationCacheEntry entry = this.sqlClassificationCache.get(index); 1017 1018 if (entry == null) { 1019 if (emptyIndex < 0) 1020 emptyIndex = index; 1021 } else if (entry.sql().equals(sql)) { 1022 return entry.classification(); 1023 } 1024 } 1025 1026 SqlAnalysis analysis = analyzeSql(sql); 1027 SqlClassification classification = new SqlClassification(analysis.operation(), 1028 this.recordCollectionName ? collectionName(analysis) : null); 1029 SqlClassificationCacheEntry newEntry = new SqlClassificationCacheEntry(sql, classification); 1030 1031 if (emptyIndex < 0 || !this.sqlClassificationCache.compareAndSet(emptyIndex, null, newEntry)) 1032 this.sqlClassificationCache.set(firstIndex, newEntry); 1033 1034 return classification; 1035 } 1036 1037 private static int spreadHash(int hash) { 1038 return hash ^ (hash >>> 16); 1039 } 1040 1041 @Nullable 1042 private static String collectionName(@NonNull SqlAnalysis analysis) { 1043 requireNonNull(analysis); 1044 if (analysis.operationIndex() < 0 || analysis.commonTableExpression() || hasSecondaryOperation(analysis)) 1045 return null; 1046 1047 return switch (analysis.operation()) { 1048 case "INSERT" -> insertCollectionName(analysis); 1049 case "UPDATE" -> updateCollectionName(analysis); 1050 case "DELETE" -> deleteCollectionName(analysis); 1051 case "SELECT" -> selectCollectionName(analysis); 1052 default -> null; 1053 }; 1054 } 1055 1056 private static boolean hasSecondaryOperation(@NonNull SqlAnalysis analysis) { 1057 for (int i = analysis.operationIndex() + 1; i < analysis.endIndex(); i++) 1058 if (operationWord(analysis.tokens().get(i))) 1059 return true; 1060 1061 return false; 1062 } 1063 1064 @Nullable 1065 private static String insertCollectionName(@NonNull SqlAnalysis analysis) { 1066 int index = analysis.operationIndex() + 1; 1067 if (!isWord(analysis.tokens(), index, "INTO")) 1068 return null; 1069 1070 ParsedIdentifier identifier = parseIdentifier(analysis.tokens(), index + 1, analysis.endIndex()); 1071 return identifier == null ? null : identifier.value(); 1072 } 1073 1074 @Nullable 1075 private static String updateCollectionName(@NonNull SqlAnalysis analysis) { 1076 int index = analysis.operationIndex() + 1; 1077 if (isWord(analysis.tokens(), index, "ONLY")) 1078 index++; 1079 1080 ParsedIdentifier identifier = parseIdentifier(analysis.tokens(), index, analysis.endIndex()); 1081 if (identifier == null) 1082 return null; 1083 1084 int depth = 0; 1085 for (int i = identifier.nextIndex(); i < analysis.endIndex(); i++) { 1086 SqlToken token = analysis.tokens().get(i); 1087 if (token.isSymbol("(")) 1088 depth++; 1089 else if (token.isSymbol(")")) 1090 depth = Math.max(0, depth - 1); 1091 else if (depth == 0 && token.isWord("SET")) 1092 return identifier.value(); 1093 else if (depth == 0 && (token.isSymbol(",") || token.isWord("JOIN"))) 1094 return null; 1095 } 1096 1097 return null; 1098 } 1099 1100 @Nullable 1101 private static String deleteCollectionName(@NonNull SqlAnalysis analysis) { 1102 int index = analysis.operationIndex() + 1; 1103 if (!isWord(analysis.tokens(), index, "FROM")) 1104 return null; 1105 1106 ParsedIdentifier identifier = parseIdentifier(analysis.tokens(), index + 1, analysis.endIndex()); 1107 if (identifier == null) 1108 return null; 1109 1110 int depth = 0; 1111 for (int i = identifier.nextIndex(); i < analysis.endIndex(); i++) { 1112 SqlToken token = analysis.tokens().get(i); 1113 if (token.isSymbol("(")) 1114 depth++; 1115 else if (token.isSymbol(")")) 1116 depth = Math.max(0, depth - 1); 1117 else if (depth == 0 && (token.isSymbol(",") || token.isWord("JOIN") || token.isWord("USING"))) 1118 return null; 1119 } 1120 1121 return identifier.value(); 1122 } 1123 1124 @Nullable 1125 private static String selectCollectionName(@NonNull SqlAnalysis analysis) { 1126 List<SqlToken> tokens = analysis.tokens(); 1127 int depth = 0; 1128 int fromIndex = -1; 1129 1130 for (int i = analysis.operationIndex() + 1; i < analysis.endIndex(); i++) { 1131 SqlToken token = tokens.get(i); 1132 if (token.isSymbol("(")) 1133 depth++; 1134 else if (token.isSymbol(")")) 1135 depth = Math.max(0, depth - 1); 1136 else if (depth == 0 && token.isWord("FROM")) { 1137 fromIndex = i; 1138 break; 1139 } 1140 } 1141 1142 if (fromIndex < 0 || fromIndex + 1 >= analysis.endIndex() 1143 || tokens.get(fromIndex + 1).isWord("LATERAL") 1144 || tokens.get(fromIndex + 1).isSymbol("(")) 1145 return null; 1146 1147 ParsedIdentifier identifier = parseIdentifier(tokens, fromIndex + 1, analysis.endIndex()); 1148 if (identifier == null || (identifier.nextIndex() < analysis.endIndex() 1149 && tokens.get(identifier.nextIndex()).isSymbol("("))) 1150 return null; 1151 1152 depth = 0; 1153 for (int i = identifier.nextIndex(); i < analysis.endIndex(); i++) { 1154 SqlToken token = tokens.get(i); 1155 if (token.isSymbol("(")) 1156 depth++; 1157 else if (token.isSymbol(")")) 1158 depth = Math.max(0, depth - 1); 1159 else if (depth == 0 && (token.isSymbol(",") || token.isWord("JOIN") || token.isWord("APPLY") 1160 || token.isWord("UNION") || token.isWord("INTERSECT") || token.isWord("EXCEPT") 1161 || token.isWord("FROM"))) 1162 return null; 1163 } 1164 1165 return identifier.value(); 1166 } 1167 1168 @Nullable 1169 private static ParsedIdentifier parseIdentifier(@NonNull List<SqlToken> tokens, 1170 int startIndex, 1171 int endIndex) { 1172 if (startIndex >= endIndex || !tokens.get(startIndex).identifier()) 1173 return null; 1174 1175 StringBuilder value = new StringBuilder(tokens.get(startIndex).text()); 1176 int index = startIndex + 1; 1177 1178 while (index + 1 < endIndex && tokens.get(index).isSymbol(".") && tokens.get(index + 1).identifier()) { 1179 value.append('.').append(tokens.get(index + 1).text()); 1180 index += 2; 1181 } 1182 1183 return new ParsedIdentifier(value.toString(), index); 1184 } 1185 1186 private static boolean isWord(@NonNull List<SqlToken> tokens, 1187 int index, 1188 @NonNull String word) { 1189 return index >= 0 && index < tokens.size() && tokens.get(index).isWord(word); 1190 } 1191 1192 @NonNull 1193 private static SqlAnalysis analyzeSql(@NonNull String sql) { 1194 List<SqlToken> tokens = tokenizeSql(requireNonNull(sql)); 1195 int endIndex = firstStatementEnd(tokens); 1196 if (endIndex == 0 || !tokens.get(0).word()) 1197 return new SqlAnalysis(tokens, -1, endIndex, UNKNOWN_OPERATION, false); 1198 1199 if (!tokens.get(0).isWord("WITH")) 1200 return new SqlAnalysis(tokens, 0, endIndex, tokens.get(0).upperText(), false); 1201 1202 int depth = 0; 1203 boolean completedCommonTableExpression = false; 1204 for (int i = 1; i < endIndex; i++) { 1205 SqlToken token = tokens.get(i); 1206 if (token.isSymbol("(")) { 1207 depth++; 1208 } else if (token.isSymbol(")")) { 1209 if (depth == 1) 1210 completedCommonTableExpression = true; 1211 depth = Math.max(0, depth - 1); 1212 } else if (depth == 0 && completedCommonTableExpression && operationWord(token)) { 1213 return new SqlAnalysis(tokens, i, endIndex, token.upperText(), true); 1214 } 1215 } 1216 1217 return new SqlAnalysis(tokens, -1, endIndex, UNKNOWN_OPERATION, true); 1218 } 1219 1220 private static int firstStatementEnd(@NonNull List<SqlToken> tokens) { 1221 int depth = 0; 1222 for (int i = 0; i < tokens.size(); i++) { 1223 SqlToken token = tokens.get(i); 1224 if (token.isSymbol("(")) 1225 depth++; 1226 else if (token.isSymbol(")")) 1227 depth = Math.max(0, depth - 1); 1228 else if (depth == 0 && token.isSymbol(";")) 1229 return i; 1230 } 1231 1232 return tokens.size(); 1233 } 1234 1235 private static boolean operationWord(@NonNull SqlToken token) { 1236 if (!token.word()) 1237 return false; 1238 1239 return switch (token.upperText()) { 1240 case "SELECT", "INSERT", "UPDATE", "DELETE", "MERGE", "CALL", "EXECUTE", "CREATE", "ALTER", 1241 "DROP", "TRUNCATE", "GRANT", "REVOKE" -> true; 1242 default -> false; 1243 }; 1244 } 1245 1246 @NonNull 1247 private static List<SqlToken> tokenizeSql(@NonNull String sql) { 1248 List<SqlToken> tokens = new ArrayList<>(); 1249 int index = 0; 1250 1251 while (index < sql.length()) { 1252 char character = sql.charAt(index); 1253 if (Character.isWhitespace(character)) { 1254 index++; 1255 continue; 1256 } 1257 1258 if (character == '-' && index + 1 < sql.length() && sql.charAt(index + 1) == '-') { 1259 index = skipLine(sql, index + 2); 1260 continue; 1261 } 1262 1263 if (character == '#') { 1264 index = skipLine(sql, index + 1); 1265 continue; 1266 } 1267 1268 if (character == '/' && index + 1 < sql.length() && sql.charAt(index + 1) == '*') { 1269 index = skipBlockComment(sql, index + 2); 1270 continue; 1271 } 1272 1273 if ((character == 'q' || character == 'Q') && index + 2 < sql.length() && sql.charAt(index + 1) == '\'') { 1274 int nextIndex = skipOracleQuotedString(sql, index); 1275 if (nextIndex > index) { 1276 index = nextIndex; 1277 continue; 1278 } 1279 } 1280 1281 if (character == '\'') { 1282 index = skipQuoted(sql, index + 1, '\'', true); 1283 continue; 1284 } 1285 1286 if (character == '$') { 1287 int nextIndex = skipDollarQuotedString(sql, index); 1288 if (nextIndex > index) { 1289 index = nextIndex; 1290 continue; 1291 } 1292 } 1293 1294 if (character == '\"' || character == '`' || character == '[') { 1295 QuotedIdentifier quotedIdentifier = readQuotedIdentifier(sql, index, character); 1296 tokens.add(new SqlToken(SqlTokenType.IDENTIFIER, quotedIdentifier.value())); 1297 index = quotedIdentifier.nextIndex(); 1298 continue; 1299 } 1300 1301 if (Character.isLetter(character) || character == '_') { 1302 int startIndex = index++; 1303 while (index < sql.length()) { 1304 char nextCharacter = sql.charAt(index); 1305 if (!Character.isLetterOrDigit(nextCharacter) && nextCharacter != '_' && nextCharacter != '$') 1306 break; 1307 index++; 1308 } 1309 tokens.add(new SqlToken(SqlTokenType.WORD, sql.substring(startIndex, index))); 1310 continue; 1311 } 1312 1313 tokens.add(new SqlToken(SqlTokenType.SYMBOL, String.valueOf(character))); 1314 index++; 1315 } 1316 1317 return List.copyOf(tokens); 1318 } 1319 1320 private static int skipLine(@NonNull String sql, 1321 int startIndex) { 1322 int newlineIndex = sql.indexOf('\n', startIndex); 1323 return newlineIndex < 0 ? sql.length() : newlineIndex + 1; 1324 } 1325 1326 private static int skipBlockComment(@NonNull String sql, 1327 int startIndex) { 1328 int depth = 1; 1329 int index = startIndex; 1330 while (index < sql.length() && depth > 0) { 1331 if (index + 1 < sql.length() && sql.charAt(index) == '/' && sql.charAt(index + 1) == '*') { 1332 depth++; 1333 index += 2; 1334 } else if (index + 1 < sql.length() && sql.charAt(index) == '*' && sql.charAt(index + 1) == '/') { 1335 depth--; 1336 index += 2; 1337 } else { 1338 index++; 1339 } 1340 } 1341 return index; 1342 } 1343 1344 private static int skipQuoted(@NonNull String sql, 1345 int startIndex, 1346 char quote, 1347 boolean backslashEscapes) { 1348 int index = startIndex; 1349 while (index < sql.length()) { 1350 char character = sql.charAt(index); 1351 if (backslashEscapes && character == '\\' && index + 1 < sql.length()) { 1352 index += 2; 1353 } else if (character == quote) { 1354 if (index + 1 < sql.length() && sql.charAt(index + 1) == quote) 1355 index += 2; 1356 else 1357 return index + 1; 1358 } else { 1359 index++; 1360 } 1361 } 1362 return sql.length(); 1363 } 1364 1365 private static int skipDollarQuotedString(@NonNull String sql, 1366 int startIndex) { 1367 int delimiterEnd = startIndex + 1; 1368 while (delimiterEnd < sql.length()) { 1369 char character = sql.charAt(delimiterEnd); 1370 if (character == '$') 1371 break; 1372 if (!Character.isLetterOrDigit(character) && character != '_') 1373 return startIndex; 1374 delimiterEnd++; 1375 } 1376 1377 if (delimiterEnd >= sql.length()) 1378 return startIndex; 1379 1380 String delimiter = sql.substring(startIndex, delimiterEnd + 1); 1381 int closingIndex = sql.indexOf(delimiter, delimiterEnd + 1); 1382 return closingIndex < 0 ? sql.length() : closingIndex + delimiter.length(); 1383 } 1384 1385 private static int skipOracleQuotedString(@NonNull String sql, 1386 int startIndex) { 1387 char openingDelimiter = sql.charAt(startIndex + 2); 1388 char closingDelimiter = switch (openingDelimiter) { 1389 case '[' -> ']'; 1390 case '(' -> ')'; 1391 case '{' -> '}'; 1392 case '<' -> '>'; 1393 default -> openingDelimiter; 1394 }; 1395 1396 for (int index = startIndex + 3; index + 1 < sql.length(); index++) { 1397 if (sql.charAt(index) == closingDelimiter && sql.charAt(index + 1) == '\'') 1398 return index + 2; 1399 } 1400 return sql.length(); 1401 } 1402 1403 @NonNull 1404 private static QuotedIdentifier readQuotedIdentifier(@NonNull String sql, 1405 int startIndex, 1406 char openingQuote) { 1407 char closingQuote = openingQuote == '[' ? ']' : openingQuote; 1408 StringBuilder value = new StringBuilder(); 1409 int index = startIndex + 1; 1410 while (index < sql.length()) { 1411 char character = sql.charAt(index); 1412 if (character == closingQuote) { 1413 if (index + 1 < sql.length() && sql.charAt(index + 1) == closingQuote) { 1414 value.append(closingQuote); 1415 index += 2; 1416 continue; 1417 } 1418 return new QuotedIdentifier(value.toString(), index + 1); 1419 } 1420 value.append(character); 1421 index++; 1422 } 1423 1424 return new QuotedIdentifier(value.toString(), sql.length()); 1425 } 1426 1427 private enum SqlTokenType { 1428 WORD, 1429 IDENTIFIER, 1430 SYMBOL 1431 } 1432 1433 private record SqlToken(@NonNull SqlTokenType type, 1434 @NonNull String text) { 1435 private boolean word() { 1436 return this.type == SqlTokenType.WORD; 1437 } 1438 1439 private boolean identifier() { 1440 return this.type == SqlTokenType.WORD || this.type == SqlTokenType.IDENTIFIER; 1441 } 1442 1443 private boolean isWord(@NonNull String word) { 1444 return word() && this.text.equalsIgnoreCase(word); 1445 } 1446 1447 private boolean isSymbol(@NonNull String symbol) { 1448 return this.type == SqlTokenType.SYMBOL && this.text.equals(symbol); 1449 } 1450 1451 @NonNull 1452 private String upperText() { 1453 return this.text.toUpperCase(Locale.ROOT); 1454 } 1455 } 1456 1457 private record SqlAnalysis(@NonNull List<SqlToken> tokens, 1458 int operationIndex, 1459 int endIndex, 1460 @NonNull String operation, 1461 boolean commonTableExpression) { 1462 } 1463 1464 private record SqlClassification(@NonNull String operation, 1465 @Nullable String collectionName) { 1466 } 1467 1468 private record SqlClassificationCacheEntry(@NonNull String sql, 1469 @NonNull SqlClassification classification) { 1470 } 1471 1472 private record ParsedIdentifier(@NonNull String value, 1473 int nextIndex) { 1474 } 1475 1476 private record QuotedIdentifier(@NonNull String value, 1477 int nextIndex) { 1478 } 1479 1480 @NonNull 1481 private static String enumValue(@NonNull Enum<?> value) { 1482 requireNonNull(value); 1483 return value.name().toLowerCase(Locale.ROOT); 1484 } 1485 1486 @Nullable 1487 private String dbResponseStatusCode(@Nullable Throwable throwable) { 1488 SQLException sqlException = sqlExceptionFor(throwable); 1489 if (sqlException == null) 1490 return null; 1491 1492 String sqlState = sqlException.getSQLState(); 1493 if (sqlState == null || sqlState.isBlank()) 1494 return null; 1495 1496 if (this.recordFullSqlState || sqlState.length() < 2) 1497 return sqlState; 1498 1499 return sqlState.substring(0, 2); 1500 } 1501 1502 @Nullable 1503 private static SQLException sqlExceptionFor(@Nullable Throwable throwable) { 1504 Throwable current = throwable; 1505 Set<Throwable> visited = Collections.newSetFromMap(new IdentityHashMap<>()); 1506 1507 while (current != null && visited.add(current)) { 1508 if (current instanceof SQLException sqlException) 1509 return sqlException; 1510 1511 current = current.getCause(); 1512 } 1513 1514 return null; 1515 } 1516 1517 @NonNull 1518 private static String errorType(@NonNull Throwable throwable) { 1519 requireNonNull(throwable); 1520 Throwable current = throwable; 1521 Throwable mostRelevant = throwable; 1522 Set<Throwable> visited = Collections.newSetFromMap(new IdentityHashMap<>()); 1523 1524 while (current != null && visited.add(current)) { 1525 if (current instanceof SQLException) 1526 return current.getClass().getName(); 1527 mostRelevant = current; 1528 current = current.getCause(); 1529 } 1530 1531 return mostRelevant.getClass().getName(); 1532 } 1533 1534 private static double seconds(@NonNull Duration duration) { 1535 requireNonNull(duration); 1536 return duration.toNanos() / 1_000_000_000D; 1537 } 1538 1539 /** 1540 * Builder used to construct instances of {@link OpenTelemetryMetricsCollector}. 1541 */ 1542 @NotThreadSafe 1543 public static final class Builder { 1544 @Nullable 1545 private Meter meter; 1546 @Nullable 1547 private OpenTelemetry openTelemetry; 1548 @NonNull 1549 private String instrumentationName; 1550 @Nullable 1551 private String instrumentationVersion; 1552 @Nullable 1553 private String poolName; 1554 @Nullable 1555 private String namespace; 1556 @NonNull 1557 private Boolean recordCollectionName; 1558 @NonNull 1559 private Boolean recordFullSqlState; 1560 1561 private Builder() { 1562 this.openTelemetry = null; 1563 this.instrumentationName = DEFAULT_INSTRUMENTATION_NAME; 1564 this.instrumentationVersion = packageImplementationVersion(); 1565 this.poolName = null; 1566 this.namespace = null; 1567 this.recordCollectionName = false; 1568 this.recordFullSqlState = true; 1569 } 1570 1571 /** 1572 * Sets a specific meter to use for metric instruments. 1573 * 1574 * @param meter the meter to use 1575 * @return this builder 1576 */ 1577 @NonNull 1578 public Builder meter(@NonNull Meter meter) { 1579 this.meter = requireNonNull(meter); 1580 return this; 1581 } 1582 1583 /** 1584 * Sets the OpenTelemetry API object used to construct a meter if {@link #meter(Meter)} is not set. 1585 * 1586 * @param openTelemetry the OpenTelemetry instance 1587 * @return this builder 1588 */ 1589 @NonNull 1590 public Builder openTelemetry(@NonNull OpenTelemetry openTelemetry) { 1591 this.openTelemetry = requireNonNull(openTelemetry); 1592 return this; 1593 } 1594 1595 /** 1596 * Sets the instrumentation scope name to use when constructing a meter. 1597 * 1598 * @param instrumentationName the instrumentation scope name 1599 * @return this builder 1600 */ 1601 @NonNull 1602 public Builder instrumentationName(@NonNull String instrumentationName) { 1603 this.instrumentationName = requireNonNull(instrumentationName); 1604 return this; 1605 } 1606 1607 /** 1608 * Sets an optional instrumentation scope version to use when constructing a meter. 1609 * 1610 * @param instrumentationVersion the instrumentation scope version, or {@code null} 1611 * @return this builder 1612 */ 1613 @NonNull 1614 public Builder instrumentationVersion(@Nullable String instrumentationVersion) { 1615 this.instrumentationVersion = instrumentationVersion; 1616 return this; 1617 } 1618 1619 /** 1620 * Sets the database connection pool name to emit as {@code db.client.connection.pool.name}. 1621 * <p> 1622 * When unset or {@code null}, {@code db.client.connection.wait_time} and 1623 * {@code db.client.connection.use_time} are not emitted, and notification metrics omit the pool-name attribute. 1624 * 1625 * @param poolName pool name, or {@code null} 1626 * @return this builder 1627 */ 1628 @NonNull 1629 public Builder poolName(@Nullable String poolName) { 1630 this.poolName = poolName; 1631 return this; 1632 } 1633 1634 /** 1635 * Sets the logical database namespace to emit as {@code db.namespace}. 1636 * <p> 1637 * No JDBC metadata lookup is performed. When unset or {@code null}, the attribute is omitted. 1638 * 1639 * @param namespace logical database namespace, or {@code null} 1640 * @return this builder 1641 */ 1642 @NonNull 1643 public Builder namespace(@Nullable String namespace) { 1644 this.namespace = namespace; 1645 return this; 1646 } 1647 1648 /** 1649 * Controls whether parsed collection/table names are emitted as {@code db.collection.name}. 1650 * <p> 1651 * The default is {@code false}. When enabled, the flag applies consistently to 1652 * {@code db.client.operation.duration} and {@code db.client.response.returned_rows}. 1653 * 1654 * @param recordCollectionName whether to record collection/table names 1655 * @return this builder 1656 */ 1657 @NonNull 1658 public Builder recordCollectionName(@NonNull Boolean recordCollectionName) { 1659 this.recordCollectionName = requireNonNull(recordCollectionName); 1660 return this; 1661 } 1662 1663 /** 1664 * Controls whether full SQLSTATE values are emitted as {@code db.response.status_code}. 1665 * <p> 1666 * The default is {@code true}. Set this to {@code false} to record only the two-character SQLSTATE class. 1667 * 1668 * @param recordFullSqlState whether to record full SQLSTATE values 1669 * @return this builder 1670 */ 1671 @NonNull 1672 public Builder recordFullSqlState(@NonNull Boolean recordFullSqlState) { 1673 this.recordFullSqlState = requireNonNull(recordFullSqlState); 1674 return this; 1675 } 1676 1677 @NonNull 1678 private Meter resolveMeter() { 1679 if (this.meter != null) 1680 return this.meter; 1681 1682 OpenTelemetry openTelemetry = this.openTelemetry == null ? GlobalOpenTelemetry.getOrNoop() : this.openTelemetry; 1683 MeterBuilder meterBuilder = openTelemetry.meterBuilder(this.instrumentationName); 1684 1685 if (this.instrumentationVersion != null) 1686 meterBuilder = meterBuilder.setInstrumentationVersion(this.instrumentationVersion); 1687 1688 return meterBuilder.build(); 1689 } 1690 1691 /** 1692 * Builds the collector. 1693 * 1694 * @return the collector instance 1695 */ 1696 @NonNull 1697 public OpenTelemetryMetricsCollector build() { 1698 return new OpenTelemetryMetricsCollector(this); 1699 } 1700 1701 @Nullable 1702 private static String packageImplementationVersion() { 1703 Package implementationPackage = OpenTelemetryMetricsCollector.class.getPackage(); 1704 return implementationPackage == null ? null : implementationPackage.getImplementationVersion(); 1705 } 1706 } 1707}