From ac010e3899a385c28141ba37c9bc6f027e8ce528 Mon Sep 17 00:00:00 2001 From: Nick Dimiduk Date: Thu, 18 Nov 2021 09:24:45 -0800 Subject: [PATCH 1/2] 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 --- .../apache/hadoop/hbase/client/HTable.java | 52 +++++++++---- .../hbase/client/RawAsyncTableImpl.java | 39 ++++++---- .../trace/TableOperationSpanBuilder.java | 73 +++++++++++++++++++ .../hbase/client/TestAsyncTableTracing.java | 54 +++++++++----- .../hbase/client/TestHTableTracing.java | 43 ++++++++--- .../hadoop/hbase/client/TestTracingBase.java | 1 + .../hbase/trace/HBaseSemanticAttributes.java | 6 ++ 7 files changed, 210 insertions(+), 58 deletions(-) diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/HTable.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/HTable.java index 2f870b7fd3b7..7b83fad4f31c 100644 --- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/HTable.java +++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/HTable.java @@ -30,6 +30,7 @@ import io.opentelemetry.api.trace.Span; import io.opentelemetry.api.trace.SpanKind; +import io.opentelemetry.api.trace.StatusCode; import io.opentelemetry.context.Scope; import java.io.IOException; import java.io.InterruptedIOException; @@ -406,7 +407,8 @@ protected Result rpcCall() throws Exception { public Result[] get(List gets) throws IOException { final Supplier supplier = new TableOperationSpanBuilder(connection) .setTableName(tableName) - .setOperation(HBaseSemanticAttributes.Operation.BATCH); + .setOperation(HBaseSemanticAttributes.Operation.BATCH) + .setContainerOperations(gets); return TraceUtil.trace(() -> { if (gets.size() == 1) { return new Result[] { get(gets.get(0)) }; @@ -433,7 +435,8 @@ public void batch(final List actions, final Object[] results) throws InterruptedException, IOException { final Supplier supplier = new TableOperationSpanBuilder(connection) .setTableName(tableName) - .setOperation(HBaseSemanticAttributes.Operation.BATCH); + .setOperation(HBaseSemanticAttributes.Operation.BATCH) + .setContainerOperations(actions); TraceUtil.traceWithIOException(() -> { int rpcTimeout = writeRpcTimeoutMs; boolean hasRead = false; @@ -473,6 +476,7 @@ public void batch(final List actions, final Object[] results, int final Span span = new TableOperationSpanBuilder(connection) .setTableName(tableName) .setOperation(HBaseSemanticAttributes.Operation.BATCH) + .setContainerOperations(actions) .build(); try (Scope ignored = span.makeCurrent()) { AsyncRequestFuture ars = multiAp.submit(task); @@ -481,6 +485,7 @@ public void batch(final List actions, final Object[] results, int TraceUtil.setError(span, ars.getErrors()); throw ars.getErrors(); } + span.setStatus(StatusCode.OK); } finally { span.end(); } @@ -512,6 +517,7 @@ public static void doBatchWithCallback(List actions, Object[] final Span span = new TableOperationSpanBuilder(connection) .setTableName(tableName) .setOperation(HBaseSemanticAttributes.Operation.BATCH) + .setContainerOperations(actions) .build(); try (Scope ignored = span.makeCurrent()) { AsyncRequestFuture ars = connection.getAsyncProcess().submit(task); @@ -551,7 +557,8 @@ protected Void rpcCall() throws Exception { public void delete(final List deletes) throws IOException { final Supplier supplier = new TableOperationSpanBuilder(connection) .setTableName(tableName) - .setOperation(HBaseSemanticAttributes.Operation.BATCH); + .setOperation(HBaseSemanticAttributes.Operation.BATCH) + .setContainerOperations(deletes); TraceUtil.traceWithIOException(() -> { Object[] results = new Object[deletes.size()]; try { @@ -600,7 +607,8 @@ protected Void rpcCall() throws Exception { public void put(final List puts) throws IOException { final Supplier supplier = new TableOperationSpanBuilder(connection) .setTableName(tableName) - .setOperation(HBaseSemanticAttributes.Operation.BATCH); + .setOperation(HBaseSemanticAttributes.Operation.BATCH) + .setContainerOperations(puts); TraceUtil.traceWithIOException(() -> { for (Put put : puts) { validatePut(put); @@ -618,7 +626,8 @@ public void put(final List puts) throws IOException { public Result mutateRow(final RowMutations rm) throws IOException { final Supplier supplier = new TableOperationSpanBuilder(connection) .setTableName(tableName) - .setOperation(HBaseSemanticAttributes.Operation.BATCH); + .setOperation(HBaseSemanticAttributes.Operation.BATCH) + .setContainerOperations(rm); return TraceUtil.trace(() -> { long nonceGroup = getNonceGroup(); long nonce = getNonce(); @@ -773,7 +782,8 @@ public boolean checkAndPut(final byte [] row, final byte [] family, final byte [ final byte [] value, final Put put) throws IOException { final Supplier supplier = new TableOperationSpanBuilder(connection) .setTableName(tableName) - .setOperation(HBaseSemanticAttributes.Operation.CHECK_AND_MUTATE); + .setOperation(HBaseSemanticAttributes.Operation.CHECK_AND_MUTATE) + .setContainerOperations(HBaseSemanticAttributes.Operation.CHECK_AND_MUTATE, HBaseSemanticAttributes.Operation.PUT); return TraceUtil.trace( () -> doCheckAndMutate(row, family, qualifier, CompareOperator.EQUAL, value, null, null, put) .isSuccess(), @@ -786,7 +796,8 @@ public boolean checkAndPut(final byte [] row, final byte [] family, final byte [ final CompareOp compareOp, final byte [] value, final Put put) throws IOException { final Supplier supplier = new TableOperationSpanBuilder(connection) .setTableName(tableName) - .setOperation(HBaseSemanticAttributes.Operation.CHECK_AND_MUTATE); + .setOperation(HBaseSemanticAttributes.Operation.CHECK_AND_MUTATE) + .setContainerOperations(HBaseSemanticAttributes.Operation.CHECK_AND_MUTATE, HBaseSemanticAttributes.Operation.PUT); return TraceUtil.trace( () -> doCheckAndMutate(row, family, qualifier, toCompareOperator(compareOp), value, null, null, put).isSuccess(), @@ -799,7 +810,8 @@ public boolean checkAndPut(final byte [] row, final byte [] family, final byte [ final CompareOperator op, final byte [] value, final Put put) throws IOException { final Supplier supplier = new TableOperationSpanBuilder(connection) .setTableName(tableName) - .setOperation(HBaseSemanticAttributes.Operation.CHECK_AND_MUTATE); + .setOperation(HBaseSemanticAttributes.Operation.CHECK_AND_MUTATE) + .setContainerOperations(HBaseSemanticAttributes.Operation.CHECK_AND_MUTATE, HBaseSemanticAttributes.Operation.PUT); return TraceUtil.trace( () -> doCheckAndMutate(row, family, qualifier, op, value, null, null, put).isSuccess(), supplier); @@ -811,7 +823,8 @@ public boolean checkAndDelete(final byte[] row, final byte[] family, final byte[ final byte[] value, final Delete delete) throws IOException { final Supplier supplier = new TableOperationSpanBuilder(connection) .setTableName(tableName) - .setOperation(HBaseSemanticAttributes.Operation.CHECK_AND_MUTATE); + .setOperation(HBaseSemanticAttributes.Operation.CHECK_AND_MUTATE) + .setContainerOperations(HBaseSemanticAttributes.Operation.CHECK_AND_MUTATE, HBaseSemanticAttributes.Operation.DELETE); return TraceUtil.trace( () -> doCheckAndMutate(row, family, qualifier, CompareOperator.EQUAL, value, null, null, delete).isSuccess(), @@ -824,7 +837,8 @@ public boolean checkAndDelete(final byte[] row, final byte[] family, final byte[ final CompareOp compareOp, final byte[] value, final Delete delete) throws IOException { final Supplier supplier = new TableOperationSpanBuilder(connection) .setTableName(tableName) - .setOperation(HBaseSemanticAttributes.Operation.CHECK_AND_MUTATE); + .setOperation(HBaseSemanticAttributes.Operation.CHECK_AND_MUTATE) + .setContainerOperations(HBaseSemanticAttributes.Operation.CHECK_AND_MUTATE, HBaseSemanticAttributes.Operation.DELETE); return TraceUtil.trace( () -> doCheckAndMutate(row, family, qualifier, toCompareOperator(compareOp), value, null, null, delete).isSuccess(), @@ -837,7 +851,8 @@ public boolean checkAndDelete(final byte[] row, final byte[] family, final byte[ final CompareOperator op, final byte[] value, final Delete delete) throws IOException { final Supplier supplier = new TableOperationSpanBuilder(connection) .setTableName(tableName) - .setOperation(HBaseSemanticAttributes.Operation.CHECK_AND_MUTATE); + .setOperation(HBaseSemanticAttributes.Operation.CHECK_AND_MUTATE) + .setContainerOperations(HBaseSemanticAttributes.Operation.CHECK_AND_MUTATE, HBaseSemanticAttributes.Operation.DELETE); return TraceUtil.trace( () -> doCheckAndMutate(row, family, qualifier, op, value, null, null, delete).isSuccess(), supplier); @@ -914,7 +929,8 @@ public boolean checkAndMutate(final byte [] row, final byte [] family, final byt final CompareOp compareOp, final byte [] value, final RowMutations rm) throws IOException { final Supplier supplier = new TableOperationSpanBuilder(connection) .setTableName(tableName) - .setOperation(HBaseSemanticAttributes.Operation.CHECK_AND_MUTATE); + .setOperation(HBaseSemanticAttributes.Operation.CHECK_AND_MUTATE) + .setContainerOperations(rm); return TraceUtil.trace( () -> doCheckAndMutate(row, family, qualifier, toCompareOperator(compareOp), value, null, null, rm).isSuccess(), @@ -927,7 +943,8 @@ public boolean checkAndMutate(final byte [] row, final byte [] family, final byt final CompareOperator op, final byte [] value, final RowMutations rm) throws IOException { final Supplier supplier = new TableOperationSpanBuilder(connection) .setTableName(tableName) - .setOperation(HBaseSemanticAttributes.Operation.CHECK_AND_MUTATE); + .setOperation(HBaseSemanticAttributes.Operation.CHECK_AND_MUTATE) + .setContainerOperations(rm); return TraceUtil.trace( () -> doCheckAndMutate(row, family, qualifier, op, value, null, null, rm).isSuccess(), supplier); @@ -937,7 +954,8 @@ public boolean checkAndMutate(final byte [] row, final byte [] family, final byt public CheckAndMutateResult checkAndMutate(CheckAndMutate checkAndMutate) throws IOException { final Supplier supplier = new TableOperationSpanBuilder(connection) .setTableName(tableName) - .setOperation(checkAndMutate); + .setOperation(checkAndMutate) + .setContainerOperations(checkAndMutate); return TraceUtil.trace(() -> { Row action = checkAndMutate.getAction(); if (action instanceof Put || action instanceof Delete || action instanceof Increment || @@ -986,7 +1004,8 @@ public List checkAndMutate(List checkAndMu throws IOException { final Supplier supplier = new TableOperationSpanBuilder(connection) .setTableName(tableName) - .setOperation(HBaseSemanticAttributes.Operation.BATCH); + .setOperation(HBaseSemanticAttributes.Operation.BATCH) + .setContainerOperations(checkAndMutates); return TraceUtil.trace(() -> { if (checkAndMutates.isEmpty()) { return Collections.emptyList(); @@ -1056,7 +1075,8 @@ public boolean exists(final Get get) throws IOException { public boolean[] exists(List gets) throws IOException { final Supplier supplier = new TableOperationSpanBuilder(connection) .setTableName(tableName) - .setOperation(HBaseSemanticAttributes.Operation.BATCH); + .setOperation(HBaseSemanticAttributes.Operation.BATCH) + .setContainerOperations(gets); return TraceUtil.trace(() -> { if (gets.isEmpty()) { return new boolean[] {}; diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/RawAsyncTableImpl.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/RawAsyncTableImpl.java index c3cc1fb36f5c..20e84cc6f739 100644 --- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/RawAsyncTableImpl.java +++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/RawAsyncTableImpl.java @@ -386,7 +386,8 @@ public CompletableFuture thenPut(Put put) { validatePut(put, conn.connConf.getMaxKeyValueSize()); preCheck(); final Supplier supplier = newTableOperationSpanBuilder() - .setOperation(HBaseSemanticAttributes.Operation.CHECK_AND_MUTATE); + .setOperation(HBaseSemanticAttributes.Operation.CHECK_AND_MUTATE) + .setContainerOperations(put); return tracedFuture( () -> RawAsyncTableImpl.this. newCaller(row, put.getPriority(), rpcTimeoutNs) .action((controller, loc, stub) -> RawAsyncTableImpl.mutate(controller, loc, stub, put, @@ -401,7 +402,8 @@ public CompletableFuture thenPut(Put put) { public CompletableFuture thenDelete(Delete delete) { preCheck(); final Supplier supplier = newTableOperationSpanBuilder() - .setOperation(HBaseSemanticAttributes.Operation.CHECK_AND_MUTATE); + .setOperation(HBaseSemanticAttributes.Operation.CHECK_AND_MUTATE) + .setContainerOperations(delete); return tracedFuture( () -> RawAsyncTableImpl.this. newCaller(row, delete.getPriority(), rpcTimeoutNs) .action((controller, loc, stub) -> RawAsyncTableImpl.mutate(controller, loc, stub, delete, @@ -417,7 +419,8 @@ public CompletableFuture thenMutate(RowMutations mutations) { preCheck(); validatePutsInRowMutations(mutations, conn.connConf.getMaxKeyValueSize()); final Supplier supplier = newTableOperationSpanBuilder() - .setOperation(HBaseSemanticAttributes.Operation.CHECK_AND_MUTATE); + .setOperation(HBaseSemanticAttributes.Operation.CHECK_AND_MUTATE) + .setContainerOperations(mutations); return tracedFuture( () -> RawAsyncTableImpl.this . newCaller(row, mutations.getMaxPriority(), rpcTimeoutNs) @@ -460,7 +463,8 @@ public CheckAndMutateWithFilterBuilder timeRange(TimeRange timeRange) { public CompletableFuture thenPut(Put put) { validatePut(put, conn.connConf.getMaxKeyValueSize()); final Supplier supplier = newTableOperationSpanBuilder() - .setOperation(HBaseSemanticAttributes.Operation.CHECK_AND_MUTATE); + .setOperation(HBaseSemanticAttributes.Operation.CHECK_AND_MUTATE) + .setContainerOperations(put); return tracedFuture( () -> RawAsyncTableImpl.this. newCaller(row, put.getPriority(), rpcTimeoutNs) .action((controller, loc, stub) -> RawAsyncTableImpl.mutate(controller, loc, @@ -475,7 +479,8 @@ public CompletableFuture thenPut(Put put) { @Override public CompletableFuture thenDelete(Delete delete) { final Supplier supplier = newTableOperationSpanBuilder() - .setOperation(HBaseSemanticAttributes.Operation.CHECK_AND_MUTATE); + .setOperation(HBaseSemanticAttributes.Operation.CHECK_AND_MUTATE) + .setContainerOperations(delete); return tracedFuture( () -> RawAsyncTableImpl.this. newCaller(row, delete.getPriority(), rpcTimeoutNs) .action((controller, loc, stub) -> RawAsyncTableImpl.mutate(controller, loc, stub, delete, @@ -490,7 +495,8 @@ public CompletableFuture thenDelete(Delete delete) { public CompletableFuture thenMutate(RowMutations mutations) { validatePutsInRowMutations(mutations, conn.connConf.getMaxKeyValueSize()); final Supplier supplier = newTableOperationSpanBuilder() - .setOperation(HBaseSemanticAttributes.Operation.CHECK_AND_MUTATE); + .setOperation(HBaseSemanticAttributes.Operation.CHECK_AND_MUTATE) + .setContainerOperations(mutations); return tracedFuture( () -> RawAsyncTableImpl.this . newCaller(row, mutations.getMaxPriority(), rpcTimeoutNs) @@ -512,7 +518,8 @@ public CheckAndMutateWithFilterBuilder checkAndMutate(byte[] row, Filter filter) @Override public CompletableFuture checkAndMutate(CheckAndMutate checkAndMutate) { final Supplier supplier = newTableOperationSpanBuilder() - .setOperation(checkAndMutate); + .setOperation(checkAndMutate) + .setContainerOperations(checkAndMutate.getAction()); return tracedFuture(() -> { if (checkAndMutate.getAction() instanceof Put || checkAndMutate.getAction() instanceof Delete || @@ -565,7 +572,8 @@ public CompletableFuture checkAndMutate(CheckAndMutate che public List> checkAndMutate(List checkAndMutates) { final Supplier supplier = newTableOperationSpanBuilder() - .setOperation(checkAndMutates); + .setOperation(checkAndMutates) + .setContainerOperations(checkAndMutates); return tracedFutures( () -> batch(checkAndMutates, rpcTimeoutNs).stream() .map(f -> f.thenApply(r -> (CheckAndMutateResult) r)).collect(toList()), @@ -621,7 +629,8 @@ public CompletableFuture mutateRow(RowMutations mutations) { long nonceGroup = conn.getNonceGenerator().getNonceGroup(); long nonce = conn.getNonceGenerator().newNonce(); final Supplier supplier = newTableOperationSpanBuilder() - .setOperation(mutations); + .setOperation(mutations) + .setContainerOperations(mutations); return tracedFuture( () -> this . newCaller(mutations.getRow(), mutations.getMaxPriority(), writeRpcTimeoutNs) @@ -694,28 +703,32 @@ public void onComplete() { @Override public List> get(List gets) { final Supplier supplier = newTableOperationSpanBuilder() - .setOperation(gets); + .setOperation(gets) + .setContainerOperations(HBaseSemanticAttributes.Operation.GET); return tracedFutures(() -> batch(gets, readRpcTimeoutNs), supplier); } @Override public List> put(List puts) { final Supplier supplier = newTableOperationSpanBuilder() - .setOperation(puts); + .setOperation(puts) + .setContainerOperations(HBaseSemanticAttributes.Operation.PUT); return tracedFutures(() -> voidMutate(puts), supplier); } @Override public List> delete(List deletes) { final Supplier supplier = newTableOperationSpanBuilder() - .setOperation(deletes); + .setOperation(deletes) + .setContainerOperations(HBaseSemanticAttributes.Operation.DELETE); return tracedFutures(() -> voidMutate(deletes), supplier); } @Override public List> batch(List actions) { final Supplier supplier = newTableOperationSpanBuilder() - .setOperation(actions); + .setOperation(actions) + .setContainerOperations(actions); return tracedFutures(() -> batch(actions, rpcTimeoutNs), supplier); } diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/trace/TableOperationSpanBuilder.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/trace/TableOperationSpanBuilder.java index 2b9314a7ee84..7db4edb1a82f 100644 --- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/trace/TableOperationSpanBuilder.java +++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/trace/TableOperationSpanBuilder.java @@ -18,15 +18,22 @@ package org.apache.hadoop.hbase.client.trace; +import static org.apache.hadoop.hbase.trace.HBaseSemanticAttributes.CONTAINER_DB_OPERATIONS_KEY; import static org.apache.hadoop.hbase.trace.HBaseSemanticAttributes.DB_OPERATION; import io.opentelemetry.api.common.AttributeKey; import io.opentelemetry.api.trace.Span; import io.opentelemetry.api.trace.SpanBuilder; import io.opentelemetry.api.trace.SpanKind; +import java.util.Arrays; import java.util.Collection; import java.util.HashMap; +import java.util.HashSet; +import java.util.List; import java.util.Map; +import java.util.Set; import java.util.function.Supplier; +import java.util.stream.Collectors; +import java.util.stream.Stream; import org.apache.hadoop.hbase.TableName; import org.apache.hadoop.hbase.client.Append; import org.apache.hadoop.hbase.client.AsyncConnectionImpl; @@ -90,6 +97,72 @@ public TableOperationSpanBuilder setOperation(final Operation operation) { return this; } + // `setContainerOperations` perform a recursive descent expansion of all the operations + // contained within the provided "batch" object. + + public TableOperationSpanBuilder setContainerOperations(final RowMutations mutations) { + final Operation[] ops = mutations.getMutations() + .stream() + .flatMap(row -> Stream.concat(Stream.of(valueFrom(row)), unpackRowOperations(row).stream())) + .toArray(Operation[]::new); + return setContainerOperations(ops); + } + + public TableOperationSpanBuilder setContainerOperations(final Row row) { + final Operation[] ops = + Stream.concat(Stream.of(valueFrom(row)), unpackRowOperations(row).stream()) + .toArray(Operation[]::new); + return setContainerOperations(ops); + } + + public TableOperationSpanBuilder setContainerOperations( + final Collection operations + ) { + final Operation[] ops = operations.stream() + .flatMap(row -> Stream.concat(Stream.of(valueFrom(row)), unpackRowOperations(row).stream())) + .toArray(Operation[]::new); + return setContainerOperations(ops); + } + + private static Set unpackRowOperations(final Row row) { + final Set ops = new HashSet<>(); + if (row instanceof CheckAndMutate) { + final CheckAndMutate cam = (CheckAndMutate) row; + ops.addAll(unpackRowOperations(cam)); + } + if (row instanceof RowMutations) { + final RowMutations mutations = (RowMutations) row; + ops.addAll(unpackRowOperations(mutations)); + } + return ops; + } + + private static Set unpackRowOperations(final CheckAndMutate cam) { + final Set ops = new HashSet<>(); + final Operation op = valueFrom(cam.getAction()); + switch (op) { + case BATCH: + case CHECK_AND_MUTATE: + ops.addAll(unpackRowOperations(cam.getAction())); + break; + default: + ops.add(op); + } + return ops; + } + + public TableOperationSpanBuilder setContainerOperations( + final Operation... operations + ) { + final List ops = Arrays.stream(operations) + .map(op -> op == null ? unknown : op.name()) + .sorted() + .distinct() + .collect(Collectors.toList()); + attributes.put(CONTAINER_DB_OPERATIONS_KEY, ops); + return this; + } + public TableOperationSpanBuilder setTableName(final TableName tableName) { this.tableName = tableName; TableSpanBuilder.populateTableNameAttributes(attributes, tableName); diff --git a/hbase-client/src/test/java/org/apache/hadoop/hbase/client/TestAsyncTableTracing.java b/hbase-client/src/test/java/org/apache/hadoop/hbase/client/TestAsyncTableTracing.java index 05a8ec3e21d4..a2e4f5ca85d4 100644 --- a/hbase-client/src/test/java/org/apache/hadoop/hbase/client/TestAsyncTableTracing.java +++ b/hbase-client/src/test/java/org/apache/hadoop/hbase/client/TestAsyncTableTracing.java @@ -17,6 +17,8 @@ */ package org.apache.hadoop.hbase.client; +import static org.apache.hadoop.hbase.client.trace.hamcrest.AttributesMatchers.containsEntryWithStringValuesOf; +import static org.apache.hadoop.hbase.client.trace.hamcrest.SpanDataMatchers.hasAttributes; import static org.apache.hadoop.hbase.client.trace.hamcrest.SpanDataMatchers.hasEnded; import static org.apache.hadoop.hbase.client.trace.hamcrest.SpanDataMatchers.hasKind; import static org.apache.hadoop.hbase.client.trace.hamcrest.SpanDataMatchers.hasName; @@ -335,7 +337,9 @@ public void testCheckAndMutateList() { .ifEquals(Bytes.toBytes("cf"), Bytes.toBytes("cq"), Bytes.toBytes("v")) .build(new Delete(Bytes.toBytes(0))))).toArray(new CompletableFuture[0])) .join(); - assertTrace("BATCH"); + assertTrace("BATCH", hasAttributes( + containsEntryWithStringValuesOf( + "db.hbase.container_operations", "CHECK_AND_MUTATE", "DELETE"))); } @Test @@ -343,7 +347,9 @@ public void testCheckAndMutateAll() { table.checkAndMutateAll(Arrays.asList(CheckAndMutate.newBuilder(Bytes.toBytes(0)) .ifEquals(Bytes.toBytes("cf"), Bytes.toBytes("cq"), Bytes.toBytes("v")) .build(new Delete(Bytes.toBytes(0))))).join(); - assertTrace("BATCH"); + assertTrace("BATCH", hasAttributes( + containsEntryWithStringValuesOf( + "db.hbase.container_operations", "CHECK_AND_MUTATE", "DELETE"))); } private void testCheckAndMutateBuilder(Row op) { @@ -428,12 +434,14 @@ public void testCheckAndMutateWithFilterBuilderThenMutations() throws IOExceptio } @Test - public void testMutateRow() throws Exception { - byte[] row = Bytes.toBytes(0); - RowMutations mutation = new RowMutations(row); - mutation.add(new Delete(row)); - table.mutateRow(mutation).get(); - assertTrace("BATCH"); + public void testMutateRow() throws IOException { + final RowMutations mutations = new RowMutations(Bytes.toBytes(0)) + .add((Mutation) new Put(Bytes.toBytes(0)) + .addColumn(Bytes.toBytes("cf"), Bytes.toBytes("cq"), Bytes.toBytes("v"))) + .add((Mutation) new Delete(Bytes.toBytes(0))); + table.mutateRow(mutations).join(); + assertTrace("BATCH", hasAttributes( + containsEntryWithStringValuesOf("db.hbase.container_operations", "DELETE", "PUT"))); } @Test @@ -448,13 +456,15 @@ public void testExistsList() { .allOf( table.exists(Arrays.asList(new Get(Bytes.toBytes(0)))).toArray(new CompletableFuture[0])) .join(); - assertTrace("BATCH"); + assertTrace("BATCH", hasAttributes( + containsEntryWithStringValuesOf("db.hbase.container_operations", "GET"))); } @Test public void testExistsAll() { table.existsAll(Arrays.asList(new Get(Bytes.toBytes(0)))).join(); - assertTrace("BATCH"); + assertTrace("BATCH", hasAttributes( + containsEntryWithStringValuesOf("db.hbase.container_operations", "GET"))); } @Test @@ -462,13 +472,15 @@ public void testGetList() { CompletableFuture .allOf(table.get(Arrays.asList(new Get(Bytes.toBytes(0)))).toArray(new CompletableFuture[0])) .join(); - assertTrace("BATCH"); + assertTrace("BATCH", hasAttributes( + containsEntryWithStringValuesOf("db.hbase.container_operations", "GET"))); } @Test public void testGetAll() { table.getAll(Arrays.asList(new Get(Bytes.toBytes(0)))).join(); - assertTrace("BATCH"); + assertTrace("BATCH", hasAttributes( + containsEntryWithStringValuesOf("db.hbase.container_operations", "GET"))); } @Test @@ -477,14 +489,16 @@ public void testPutList() { .allOf(table.put(Arrays.asList(new Put(Bytes.toBytes(0)).addColumn(Bytes.toBytes("cf"), Bytes.toBytes("cq"), Bytes.toBytes("v")))).toArray(new CompletableFuture[0])) .join(); - assertTrace("BATCH"); + assertTrace("BATCH", hasAttributes( + containsEntryWithStringValuesOf("db.hbase.container_operations", "PUT"))); } @Test public void testPutAll() { table.putAll(Arrays.asList(new Put(Bytes.toBytes(0)).addColumn(Bytes.toBytes("cf"), Bytes.toBytes("cq"), Bytes.toBytes("v")))).join(); - assertTrace("BATCH"); + assertTrace("BATCH", hasAttributes( + containsEntryWithStringValuesOf("db.hbase.container_operations", "PUT"))); } @Test @@ -493,13 +507,15 @@ public void testDeleteList() { .allOf( table.delete(Arrays.asList(new Delete(Bytes.toBytes(0)))).toArray(new CompletableFuture[0])) .join(); - assertTrace("BATCH"); + assertTrace("BATCH", hasAttributes( + containsEntryWithStringValuesOf("db.hbase.container_operations", "DELETE"))); } @Test public void testDeleteAll() { table.deleteAll(Arrays.asList(new Delete(Bytes.toBytes(0)))).join(); - assertTrace("BATCH"); + assertTrace("BATCH", hasAttributes( + containsEntryWithStringValuesOf("db.hbase.container_operations", "DELETE"))); } @Test @@ -508,12 +524,14 @@ public void testBatch() { .allOf( table.batch(Arrays.asList(new Delete(Bytes.toBytes(0)))).toArray(new CompletableFuture[0])) .join(); - assertTrace("BATCH"); + assertTrace("BATCH", hasAttributes( + containsEntryWithStringValuesOf("db.hbase.container_operations", "DELETE"))); } @Test public void testBatchAll() { table.batchAll(Arrays.asList(new Delete(Bytes.toBytes(0)))).join(); - assertTrace("BATCH"); + assertTrace("BATCH", hasAttributes( + containsEntryWithStringValuesOf("db.hbase.container_operations", "DELETE"))); } } diff --git a/hbase-client/src/test/java/org/apache/hadoop/hbase/client/TestHTableTracing.java b/hbase-client/src/test/java/org/apache/hadoop/hbase/client/TestHTableTracing.java index 80db9a122e37..4b94ad9f36de 100644 --- a/hbase-client/src/test/java/org/apache/hadoop/hbase/client/TestHTableTracing.java +++ b/hbase-client/src/test/java/org/apache/hadoop/hbase/client/TestHTableTracing.java @@ -17,6 +17,8 @@ */ package org.apache.hadoop.hbase.client; +import static org.apache.hadoop.hbase.client.trace.hamcrest.AttributesMatchers.containsEntryWithStringValuesOf; +import static org.apache.hadoop.hbase.client.trace.hamcrest.SpanDataMatchers.hasAttributes; import static org.apache.hadoop.hbase.client.trace.hamcrest.SpanDataMatchers.hasEnded; import static org.apache.hadoop.hbase.client.trace.hamcrest.SpanDataMatchers.hasKind; import static org.apache.hadoop.hbase.client.trace.hamcrest.SpanDataMatchers.hasName; @@ -320,7 +322,9 @@ public void testCheckAndMutate() throws IOException { table.checkAndMutate(CheckAndMutate.newBuilder(Bytes.toBytes(0)) .ifEquals(Bytes.toBytes("cf"), Bytes.toBytes("cq"), Bytes.toBytes("v")) .build(new Delete(Bytes.toBytes(0)))); - assertTrace("CHECK_AND_MUTATE"); + assertTrace("CHECK_AND_MUTATE", hasAttributes( + containsEntryWithStringValuesOf( + "db.hbase.container_operations", "CHECK_AND_MUTATE", "DELETE"))); } @Test @@ -328,7 +332,9 @@ public void testCheckAndMutateList() throws IOException { table.checkAndMutate(Arrays.asList(CheckAndMutate.newBuilder(Bytes.toBytes(0)) .ifEquals(Bytes.toBytes("cf"), Bytes.toBytes("cq"), Bytes.toBytes("v")) .build(new Delete(Bytes.toBytes(0))))); - assertTrace("BATCH"); + assertTrace("BATCH", hasAttributes( + containsEntryWithStringValuesOf( + "db.hbase.container_operations", "CHECK_AND_MUTATE", "DELETE"))); } @Test @@ -336,51 +342,67 @@ public void testCheckAndMutateAll() throws IOException { table.checkAndMutate(Arrays.asList(CheckAndMutate.newBuilder(Bytes.toBytes(0)) .ifEquals(Bytes.toBytes("cf"), Bytes.toBytes("cq"), Bytes.toBytes("v")) .build(new Delete(Bytes.toBytes(0))))); - assertTrace("BATCH"); + assertTrace("BATCH", hasAttributes( + containsEntryWithStringValuesOf( + "db.hbase.container_operations", "CHECK_AND_MUTATE", "DELETE"))); } @Test public void testMutateRow() throws Exception { byte[] row = Bytes.toBytes(0); table.mutateRow(RowMutations.of(Arrays.asList(new Delete(row)))); - assertTrace("BATCH"); + assertTrace("BATCH", hasAttributes( + containsEntryWithStringValuesOf( + "db.hbase.container_operations", "DELETE"))); } @Test public void testExistsList() throws IOException { table.exists(Arrays.asList(new Get(Bytes.toBytes(0)))); - assertTrace("BATCH"); + assertTrace("BATCH", hasAttributes( + containsEntryWithStringValuesOf( + "db.hbase.container_operations", "GET"))); } @Test public void testExistsAll() throws IOException { table.existsAll(Arrays.asList(new Get(Bytes.toBytes(0)))); - assertTrace("BATCH"); + assertTrace("BATCH", hasAttributes( + containsEntryWithStringValuesOf( + "db.hbase.container_operations", "GET"))); } @Test public void testGetList() throws IOException { table.get(Arrays.asList(new Get(Bytes.toBytes(0)))); - assertTrace("BATCH"); + assertTrace("BATCH", hasAttributes( + containsEntryWithStringValuesOf( + "db.hbase.container_operations", "GET"))); } @Test public void testPutList() throws IOException { table.put(Arrays.asList(new Put(Bytes.toBytes(0)).addColumn(Bytes.toBytes("cf"), Bytes.toBytes("cq"), Bytes.toBytes("v")))); - assertTrace("BATCH"); + assertTrace("BATCH", hasAttributes( + containsEntryWithStringValuesOf( + "db.hbase.container_operations", "PUT"))); } @Test public void testDeleteList() throws IOException { table.delete(Lists.newArrayList(new Delete(Bytes.toBytes(0)))); - assertTrace("BATCH"); + assertTrace("BATCH", hasAttributes( + containsEntryWithStringValuesOf( + "db.hbase.container_operations", "DELETE"))); } @Test public void testBatchList() throws IOException, InterruptedException { table.batch(Arrays.asList(new Delete(Bytes.toBytes(0))), null); - assertTrace("BATCH"); + assertTrace("BATCH", hasAttributes( + containsEntryWithStringValuesOf( + "db.hbase.container_operations", "DELETE"))); } @Test @@ -388,5 +410,4 @@ public void testTableClose() throws IOException { table.close(); assertTrace(HTable.class.getSimpleName(), "close", null, TableName.META_TABLE_NAME); } - } diff --git a/hbase-client/src/test/java/org/apache/hadoop/hbase/client/TestTracingBase.java b/hbase-client/src/test/java/org/apache/hadoop/hbase/client/TestTracingBase.java index 34633481c210..2a10d3b9e8c6 100644 --- a/hbase-client/src/test/java/org/apache/hadoop/hbase/client/TestTracingBase.java +++ b/hbase-client/src/test/java/org/apache/hadoop/hbase/client/TestTracingBase.java @@ -63,6 +63,7 @@ public void setUp() throws Exception { conf = HBaseConfiguration.create(); conf.set(HConstants.CLIENT_CONNECTION_REGISTRY_IMPL_CONF_KEY, RegistryForTracingTest.class.getName()); + TRACE_RULE.clearSpans(); } protected void assertTrace(String className, String methodName, ServerName serverName, diff --git a/hbase-common/src/main/java/org/apache/hadoop/hbase/trace/HBaseSemanticAttributes.java b/hbase-common/src/main/java/org/apache/hadoop/hbase/trace/HBaseSemanticAttributes.java index fd6ab852e063..1a74fdcd65a2 100644 --- a/hbase-common/src/main/java/org/apache/hadoop/hbase/trace/HBaseSemanticAttributes.java +++ b/hbase-common/src/main/java/org/apache/hadoop/hbase/trace/HBaseSemanticAttributes.java @@ -36,6 +36,12 @@ public final class HBaseSemanticAttributes { public static final AttributeKey DB_NAME = SemanticAttributes.DB_NAME; public static final AttributeKey DB_OPERATION = SemanticAttributes.DB_OPERATION; public static final AttributeKey TABLE_KEY = AttributeKey.stringKey("db.hbase.table"); + /** + * For operations that themselves ship one or more operations, such as + * {@link Operation#BATCH} and {@link Operation#CHECK_AND_MUTATE}. + */ + public static final AttributeKey> CONTAINER_DB_OPERATIONS_KEY = + AttributeKey.stringArrayKey("db.hbase.container_operations"); public static final AttributeKey> REGION_NAMES_KEY = AttributeKey.stringArrayKey("db.hbase.regions"); public static final AttributeKey RPC_SERVICE_KEY = From c3a16696d4c18ebbd7f58b0953156808925f3c67 Mon Sep 17 00:00:00 2001 From: Nick Dimiduk Date: Fri, 28 Jan 2022 16:40:49 -0800 Subject: [PATCH 2/2] Fix StackOverflowException in `TableOperationSpanBuilder` --- .../hbase/client/trace/TableOperationSpanBuilder.java | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/trace/TableOperationSpanBuilder.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/trace/TableOperationSpanBuilder.java index 7db4edb1a82f..1e8f992c7a4a 100644 --- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/trace/TableOperationSpanBuilder.java +++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/trace/TableOperationSpanBuilder.java @@ -132,7 +132,11 @@ private static Set unpackRowOperations(final Row row) { } if (row instanceof RowMutations) { final RowMutations mutations = (RowMutations) row; - ops.addAll(unpackRowOperations(mutations)); + final List operations = mutations.getMutations() + .stream() + .map(TableOperationSpanBuilder::valueFrom) + .collect(Collectors.toList()); + ops.addAll(operations); } return ops; }