Skip to content

Commit 4748707

Browse files
ndimidukndimiduk
committed
HBASE-26473 Introduce db.hbase.container_operations span attribute
For batch operations, collect and annotate the associated span with the set of all operations contained in the batch. Signed-off-by: Duo Zhang <[email protected]>
1 parent a8b68c9 commit 4748707

File tree

7 files changed

+210
-58
lines changed

7 files changed

+210
-58
lines changed

‎hbase-client/src/main/java/org/apache/hadoop/hbase/client/HTable.java‎

Lines changed: 36 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@
3030

3131
import io.opentelemetry.api.trace.Span;
3232
import io.opentelemetry.api.trace.SpanKind;
33+
import io.opentelemetry.api.trace.StatusCode;
3334
import io.opentelemetry.context.Scope;
3435
import java.io.IOException;
3536
import java.io.InterruptedIOException;
@@ -406,7 +407,8 @@ protected Result rpcCall() throws Exception {
406407
public Result[] get(List<Get> gets) throws IOException {
407408
final Supplier<Span> supplier = new TableOperationSpanBuilder(connection)
408409
.setTableName(tableName)
409-
.setOperation(HBaseSemanticAttributes.Operation.BATCH);
410+
.setOperation(HBaseSemanticAttributes.Operation.BATCH)
411+
.setContainerOperations(gets);
410412
return TraceUtil.trace(() -> {
411413
if (gets.size() == 1) {
412414
return new Result[] { get(gets.get(0)) };
@@ -433,7 +435,8 @@ public void batch(final List<? extends Row> actions, final Object[] results)
433435
throws InterruptedException, IOException {
434436
final Supplier<Span> supplier = new TableOperationSpanBuilder(connection)
435437
.setTableName(tableName)
436-
.setOperation(HBaseSemanticAttributes.Operation.BATCH);
438+
.setOperation(HBaseSemanticAttributes.Operation.BATCH)
439+
.setContainerOperations(actions);
437440
TraceUtil.traceWithIOException(() -> {
438441
int rpcTimeout = writeRpcTimeoutMs;
439442
boolean hasRead = false;
@@ -473,6 +476,7 @@ public void batch(final List<? extends Row> actions, final Object[] results, int
473476
final Span span = new TableOperationSpanBuilder(connection)
474477
.setTableName(tableName)
475478
.setOperation(HBaseSemanticAttributes.Operation.BATCH)
479+
.setContainerOperations(actions)
476480
.build();
477481
try (Scope ignored = span.makeCurrent()) {
478482
AsyncRequestFuture ars = multiAp.submit(task);
@@ -481,6 +485,7 @@ public void batch(final List<? extends Row> actions, final Object[] results, int
481485
TraceUtil.setError(span, ars.getErrors());
482486
throw ars.getErrors();
483487
}
488+
span.setStatus(StatusCode.OK);
484489
} finally {
485490
span.end();
486491
}
@@ -512,6 +517,7 @@ public static <R> void doBatchWithCallback(List<? extends Row> actions, Object[]
512517
final Span span = new TableOperationSpanBuilder(connection)
513518
.setTableName(tableName)
514519
.setOperation(HBaseSemanticAttributes.Operation.BATCH)
520+
.setContainerOperations(actions)
515521
.build();
516522
try (Scope ignored = span.makeCurrent()) {
517523
AsyncRequestFuture ars = connection.getAsyncProcess().submit(task);
@@ -551,7 +557,8 @@ protected Void rpcCall() throws Exception {
551557
public void delete(final List<Delete> deletes) throws IOException {
552558
final Supplier<Span> supplier = new TableOperationSpanBuilder(connection)
553559
.setTableName(tableName)
554-
.setOperation(HBaseSemanticAttributes.Operation.BATCH);
560+
.setOperation(HBaseSemanticAttributes.Operation.BATCH)
561+
.setContainerOperations(deletes);
555562
TraceUtil.traceWithIOException(() -> {
556563
Object[] results = new Object[deletes.size()];
557564
try {
@@ -600,7 +607,8 @@ protected Void rpcCall() throws Exception {
600607
public void put(final List<Put> puts) throws IOException {
601608
final Supplier<Span> supplier = new TableOperationSpanBuilder(connection)
602609
.setTableName(tableName)
603-
.setOperation(HBaseSemanticAttributes.Operation.BATCH);
610+
.setOperation(HBaseSemanticAttributes.Operation.BATCH)
611+
.setContainerOperations(puts);
604612
TraceUtil.traceWithIOException(() -> {
605613
for (Put put : puts) {
606614
validatePut(put);
@@ -618,7 +626,8 @@ public void put(final List<Put> puts) throws IOException {
618626
public Result mutateRow(final RowMutations rm) throws IOException {
619627
final Supplier<Span> supplier = new TableOperationSpanBuilder(connection)
620628
.setTableName(tableName)
621-
.setOperation(HBaseSemanticAttributes.Operation.BATCH);
629+
.setOperation(HBaseSemanticAttributes.Operation.BATCH)
630+
.setContainerOperations(rm);
622631
return TraceUtil.trace(() -> {
623632
long nonceGroup = getNonceGroup();
624633
long nonce = getNonce();
@@ -773,7 +782,8 @@ public boolean checkAndPut(final byte [] row, final byte [] family, final byte [
773782
final byte [] value, final Put put) throws IOException {
774783
final Supplier<Span> supplier = new TableOperationSpanBuilder(connection)
775784
.setTableName(tableName)
776-
.setOperation(HBaseSemanticAttributes.Operation.CHECK_AND_MUTATE);
785+
.setOperation(HBaseSemanticAttributes.Operation.CHECK_AND_MUTATE)
786+
.setContainerOperations(HBaseSemanticAttributes.Operation.CHECK_AND_MUTATE, HBaseSemanticAttributes.Operation.PUT);
777787
return TraceUtil.trace(
778788
() -> doCheckAndMutate(row, family, qualifier, CompareOperator.EQUAL, value, null, null, put)
779789
.isSuccess(),
@@ -786,7 +796,8 @@ public boolean checkAndPut(final byte [] row, final byte [] family, final byte [
786796
final CompareOp compareOp, final byte [] value, final Put put) throws IOException {
787797
final Supplier<Span> supplier = new TableOperationSpanBuilder(connection)
788798
.setTableName(tableName)
789-
.setOperation(HBaseSemanticAttributes.Operation.CHECK_AND_MUTATE);
799+
.setOperation(HBaseSemanticAttributes.Operation.CHECK_AND_MUTATE)
800+
.setContainerOperations(HBaseSemanticAttributes.Operation.CHECK_AND_MUTATE, HBaseSemanticAttributes.Operation.PUT);
790801
return TraceUtil.trace(
791802
() -> doCheckAndMutate(row, family, qualifier, toCompareOperator(compareOp), value, null,
792803
null, put).isSuccess(),
@@ -799,7 +810,8 @@ public boolean checkAndPut(final byte [] row, final byte [] family, final byte [
799810
final CompareOperator op, final byte [] value, final Put put) throws IOException {
800811
final Supplier<Span> supplier = new TableOperationSpanBuilder(connection)
801812
.setTableName(tableName)
802-
.setOperation(HBaseSemanticAttributes.Operation.CHECK_AND_MUTATE);
813+
.setOperation(HBaseSemanticAttributes.Operation.CHECK_AND_MUTATE)
814+
.setContainerOperations(HBaseSemanticAttributes.Operation.CHECK_AND_MUTATE, HBaseSemanticAttributes.Operation.PUT);
803815
return TraceUtil.trace(
804816
() -> doCheckAndMutate(row, family, qualifier, op, value, null, null, put).isSuccess(),
805817
supplier);
@@ -811,7 +823,8 @@ public boolean checkAndDelete(final byte[] row, final byte[] family, final byte[
811823
final byte[] value, final Delete delete) throws IOException {
812824
final Supplier<Span> supplier = new TableOperationSpanBuilder(connection)
813825
.setTableName(tableName)
814-
.setOperation(HBaseSemanticAttributes.Operation.CHECK_AND_MUTATE);
826+
.setOperation(HBaseSemanticAttributes.Operation.CHECK_AND_MUTATE)
827+
.setContainerOperations(HBaseSemanticAttributes.Operation.CHECK_AND_MUTATE, HBaseSemanticAttributes.Operation.DELETE);
815828
return TraceUtil.trace(
816829
() -> doCheckAndMutate(row, family, qualifier, CompareOperator.EQUAL, value, null, null,
817830
delete).isSuccess(),
@@ -824,7 +837,8 @@ public boolean checkAndDelete(final byte[] row, final byte[] family, final byte[
824837
final CompareOp compareOp, final byte[] value, final Delete delete) throws IOException {
825838
final Supplier<Span> supplier = new TableOperationSpanBuilder(connection)
826839
.setTableName(tableName)
827-
.setOperation(HBaseSemanticAttributes.Operation.CHECK_AND_MUTATE);
840+
.setOperation(HBaseSemanticAttributes.Operation.CHECK_AND_MUTATE)
841+
.setContainerOperations(HBaseSemanticAttributes.Operation.CHECK_AND_MUTATE, HBaseSemanticAttributes.Operation.DELETE);
828842
return TraceUtil.trace(
829843
() -> doCheckAndMutate(row, family, qualifier, toCompareOperator(compareOp), value, null,
830844
null, delete).isSuccess(),
@@ -837,7 +851,8 @@ public boolean checkAndDelete(final byte[] row, final byte[] family, final byte[
837851
final CompareOperator op, final byte[] value, final Delete delete) throws IOException {
838852
final Supplier<Span> supplier = new TableOperationSpanBuilder(connection)
839853
.setTableName(tableName)
840-
.setOperation(HBaseSemanticAttributes.Operation.CHECK_AND_MUTATE);
854+
.setOperation(HBaseSemanticAttributes.Operation.CHECK_AND_MUTATE)
855+
.setContainerOperations(HBaseSemanticAttributes.Operation.CHECK_AND_MUTATE, HBaseSemanticAttributes.Operation.DELETE);
841856
return TraceUtil.trace(
842857
() -> doCheckAndMutate(row, family, qualifier, op, value, null, null, delete).isSuccess(),
843858
supplier);
@@ -914,7 +929,8 @@ public boolean checkAndMutate(final byte [] row, final byte [] family, final byt
914929
final CompareOp compareOp, final byte [] value, final RowMutations rm) throws IOException {
915930
final Supplier<Span> supplier = new TableOperationSpanBuilder(connection)
916931
.setTableName(tableName)
917-
.setOperation(HBaseSemanticAttributes.Operation.CHECK_AND_MUTATE);
932+
.setOperation(HBaseSemanticAttributes.Operation.CHECK_AND_MUTATE)
933+
.setContainerOperations(rm);
918934
return TraceUtil.trace(
919935
() -> doCheckAndMutate(row, family, qualifier, toCompareOperator(compareOp), value, null,
920936
null, rm).isSuccess(),
@@ -927,7 +943,8 @@ public boolean checkAndMutate(final byte [] row, final byte [] family, final byt
927943
final CompareOperator op, final byte [] value, final RowMutations rm) throws IOException {
928944
final Supplier<Span> supplier = new TableOperationSpanBuilder(connection)
929945
.setTableName(tableName)
930-
.setOperation(HBaseSemanticAttributes.Operation.CHECK_AND_MUTATE);
946+
.setOperation(HBaseSemanticAttributes.Operation.CHECK_AND_MUTATE)
947+
.setContainerOperations(rm);
931948
return TraceUtil.trace(
932949
() -> doCheckAndMutate(row, family, qualifier, op, value, null, null, rm).isSuccess(),
933950
supplier);
@@ -937,7 +954,8 @@ public boolean checkAndMutate(final byte [] row, final byte [] family, final byt
937954
public CheckAndMutateResult checkAndMutate(CheckAndMutate checkAndMutate) throws IOException {
938955
final Supplier<Span> supplier = new TableOperationSpanBuilder(connection)
939956
.setTableName(tableName)
940-
.setOperation(checkAndMutate);
957+
.setOperation(checkAndMutate)
958+
.setContainerOperations(checkAndMutate);
941959
return TraceUtil.trace(() -> {
942960
Row action = checkAndMutate.getAction();
943961
if (action instanceof Put || action instanceof Delete || action instanceof Increment ||
@@ -986,7 +1004,8 @@ public List<CheckAndMutateResult> checkAndMutate(List<CheckAndMutate> checkAndMu
9861004
throws IOException {
9871005
final Supplier<Span> supplier = new TableOperationSpanBuilder(connection)
9881006
.setTableName(tableName)
989-
.setOperation(HBaseSemanticAttributes.Operation.BATCH);
1007+
.setOperation(HBaseSemanticAttributes.Operation.BATCH)
1008+
.setContainerOperations(checkAndMutates);
9901009
return TraceUtil.trace(() -> {
9911010
if (checkAndMutates.isEmpty()) {
9921011
return Collections.emptyList();
@@ -1056,7 +1075,8 @@ public boolean exists(final Get get) throws IOException {
10561075
public boolean[] exists(List<Get> gets) throws IOException {
10571076
final Supplier<Span> supplier = new TableOperationSpanBuilder(connection)
10581077
.setTableName(tableName)
1059-
.setOperation(HBaseSemanticAttributes.Operation.BATCH);
1078+
.setOperation(HBaseSemanticAttributes.Operation.BATCH)
1079+
.setContainerOperations(gets);
10601080
return TraceUtil.trace(() -> {
10611081
if (gets.isEmpty()) {
10621082
return new boolean[] {};

‎hbase-client/src/main/java/org/apache/hadoop/hbase/client/RawAsyncTableImpl.java‎

Lines changed: 26 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -386,7 +386,8 @@ public CompletableFuture<Boolean> thenPut(Put put) {
386386
validatePut(put, conn.connConf.getMaxKeyValueSize());
387387
preCheck();
388388
final Supplier<Span> supplier = newTableOperationSpanBuilder()
389-
.setOperation(HBaseSemanticAttributes.Operation.CHECK_AND_MUTATE);
389+
.setOperation(HBaseSemanticAttributes.Operation.CHECK_AND_MUTATE)
390+
.setContainerOperations(put);
390391
return tracedFuture(
391392
() -> RawAsyncTableImpl.this.<Boolean> newCaller(row, put.getPriority(), rpcTimeoutNs)
392393
.action((controller, loc, stub) -> RawAsyncTableImpl.mutate(controller, loc, stub, put,
@@ -401,7 +402,8 @@ public CompletableFuture<Boolean> thenPut(Put put) {
401402
public CompletableFuture<Boolean> thenDelete(Delete delete) {
402403
preCheck();
403404
final Supplier<Span> supplier = newTableOperationSpanBuilder()
404-
.setOperation(HBaseSemanticAttributes.Operation.CHECK_AND_MUTATE);
405+
.setOperation(HBaseSemanticAttributes.Operation.CHECK_AND_MUTATE)
406+
.setContainerOperations(delete);
405407
return tracedFuture(
406408
() -> RawAsyncTableImpl.this.<Boolean> newCaller(row, delete.getPriority(), rpcTimeoutNs)
407409
.action((controller, loc, stub) -> RawAsyncTableImpl.mutate(controller, loc, stub, delete,
@@ -417,7 +419,8 @@ public CompletableFuture<Boolean> thenMutate(RowMutations mutations) {
417419
preCheck();
418420
validatePutsInRowMutations(mutations, conn.connConf.getMaxKeyValueSize());
419421
final Supplier<Span> supplier = newTableOperationSpanBuilder()
420-
.setOperation(HBaseSemanticAttributes.Operation.CHECK_AND_MUTATE);
422+
.setOperation(HBaseSemanticAttributes.Operation.CHECK_AND_MUTATE)
423+
.setContainerOperations(mutations);
421424
return tracedFuture(
422425
() -> RawAsyncTableImpl.this
423426
.<Boolean> newCaller(row, mutations.getMaxPriority(), rpcTimeoutNs)
@@ -460,7 +463,8 @@ public CheckAndMutateWithFilterBuilder timeRange(TimeRange timeRange) {
460463
public CompletableFuture<Boolean> thenPut(Put put) {
461464
validatePut(put, conn.connConf.getMaxKeyValueSize());
462465
final Supplier<Span> supplier = newTableOperationSpanBuilder()
463-
.setOperation(HBaseSemanticAttributes.Operation.CHECK_AND_MUTATE);
466+
.setOperation(HBaseSemanticAttributes.Operation.CHECK_AND_MUTATE)
467+
.setContainerOperations(put);
464468
return tracedFuture(
465469
() -> RawAsyncTableImpl.this.<Boolean> newCaller(row, put.getPriority(), rpcTimeoutNs)
466470
.action((controller, loc, stub) -> RawAsyncTableImpl.mutate(controller, loc,
@@ -475,7 +479,8 @@ public CompletableFuture<Boolean> thenPut(Put put) {
475479
@Override
476480
public CompletableFuture<Boolean> thenDelete(Delete delete) {
477481
final Supplier<Span> supplier = newTableOperationSpanBuilder()
478-
.setOperation(HBaseSemanticAttributes.Operation.CHECK_AND_MUTATE);
482+
.setOperation(HBaseSemanticAttributes.Operation.CHECK_AND_MUTATE)
483+
.setContainerOperations(delete);
479484
return tracedFuture(
480485
() -> RawAsyncTableImpl.this.<Boolean> newCaller(row, delete.getPriority(), rpcTimeoutNs)
481486
.action((controller, loc, stub) -> RawAsyncTableImpl.mutate(controller, loc, stub, delete,
@@ -490,7 +495,8 @@ public CompletableFuture<Boolean> thenDelete(Delete delete) {
490495
public CompletableFuture<Boolean> thenMutate(RowMutations mutations) {
491496
validatePutsInRowMutations(mutations, conn.connConf.getMaxKeyValueSize());
492497
final Supplier<Span> supplier = newTableOperationSpanBuilder()
493-
.setOperation(HBaseSemanticAttributes.Operation.CHECK_AND_MUTATE);
498+
.setOperation(HBaseSemanticAttributes.Operation.CHECK_AND_MUTATE)
499+
.setContainerOperations(mutations);
494500
return tracedFuture(
495501
() -> RawAsyncTableImpl.this
496502
.<Boolean> newCaller(row, mutations.getMaxPriority(), rpcTimeoutNs)
@@ -512,7 +518,8 @@ public CheckAndMutateWithFilterBuilder checkAndMutate(byte[] row, Filter filter)
512518
@Override
513519
public CompletableFuture<CheckAndMutateResult> checkAndMutate(CheckAndMutate checkAndMutate) {
514520
final Supplier<Span> supplier = newTableOperationSpanBuilder()
515-
.setOperation(checkAndMutate);
521+
.setOperation(checkAndMutate)
522+
.setContainerOperations(checkAndMutate.getAction());
516523
return tracedFuture(() -> {
517524
if (checkAndMutate.getAction() instanceof Put ||
518525
checkAndMutate.getAction() instanceof Delete ||
@@ -565,7 +572,8 @@ public CompletableFuture<CheckAndMutateResult> checkAndMutate(CheckAndMutate che
565572
public List<CompletableFuture<CheckAndMutateResult>>
566573
checkAndMutate(List<CheckAndMutate> checkAndMutates) {
567574
final Supplier<Span> supplier = newTableOperationSpanBuilder()
568-
.setOperation(checkAndMutates);
575+
.setOperation(checkAndMutates)
576+
.setContainerOperations(checkAndMutates);
569577
return tracedFutures(
570578
() -> batch(checkAndMutates, rpcTimeoutNs).stream()
571579
.map(f -> f.thenApply(r -> (CheckAndMutateResult) r)).collect(toList()),
@@ -621,7 +629,8 @@ public CompletableFuture<Result> mutateRow(RowMutations mutations) {
621629
long nonceGroup = conn.getNonceGenerator().getNonceGroup();
622630
long nonce = conn.getNonceGenerator().newNonce();
623631
final Supplier<Span> supplier = newTableOperationSpanBuilder()
624-
.setOperation(mutations);
632+
.setOperation(mutations)
633+
.setContainerOperations(mutations);
625634
return tracedFuture(
626635
() -> this
627636
.<Result> newCaller(mutations.getRow(), mutations.getMaxPriority(), writeRpcTimeoutNs)
@@ -694,28 +703,32 @@ public void onComplete() {
694703
@Override
695704
public List<CompletableFuture<Result>> get(List<Get> gets) {
696705
final Supplier<Span> supplier = newTableOperationSpanBuilder()
697-
.setOperation(gets);
706+
.setOperation(gets)
707+
.setContainerOperations(HBaseSemanticAttributes.Operation.GET);
698708
return tracedFutures(() -> batch(gets, readRpcTimeoutNs), supplier);
699709
}
700710

701711
@Override
702712
public List<CompletableFuture<Void>> put(List<Put> puts) {
703713
final Supplier<Span> supplier = newTableOperationSpanBuilder()
704-
.setOperation(puts);
714+
.setOperation(puts)
715+
.setContainerOperations(HBaseSemanticAttributes.Operation.PUT);
705716
return tracedFutures(() -> voidMutate(puts), supplier);
706717
}
707718

708719
@Override
709720
public List<CompletableFuture<Void>> delete(List<Delete> deletes) {
710721
final Supplier<Span> supplier = newTableOperationSpanBuilder()
711-
.setOperation(deletes);
722+
.setOperation(deletes)
723+
.setContainerOperations(HBaseSemanticAttributes.Operation.DELETE);
712724
return tracedFutures(() -> voidMutate(deletes), supplier);
713725
}
714726

715727
@Override
716728
public <T> List<CompletableFuture<T>> batch(List<? extends Row> actions) {
717729
final Supplier<Span> supplier = newTableOperationSpanBuilder()
718-
.setOperation(actions);
730+
.setOperation(actions)
731+
.setContainerOperations(actions);
719732
return tracedFutures(() -> batch(actions, rpcTimeoutNs), supplier);
720733
}
721734

0 commit comments

Comments
 (0)