Skip to content

Commit fe7bc4e

Browse files
author
Colm Dougan
committed
HDDS-14004. EventNotification: Capture data to the completd operation ledger table
1 parent 7c6f3fd commit fe7bc4e

3 files changed

Lines changed: 359 additions & 2 deletions

File tree

‎hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ratis/OzoneManagerStateMachine.java‎

Lines changed: 8 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -104,6 +104,7 @@ public class OzoneManagerStateMachine extends BaseStateMachine {
104104
private final boolean isTracingEnabled;
105105
private final AtomicInteger statePausedCount = new AtomicInteger(0);
106106
private final String threadPrefix;
107+
private final OzoneManagerSuccessfulRequestHandler successfulRequestHandler;
107108

108109
/** The last {@link TermIndex} received from {@link #notifyTermIndexUpdated(long, long)}. */
109110
private volatile TermIndex lastNotifiedTermIndex = TermIndex.valueOf(0, RaftLog.INVALID_LOG_INDEX);
@@ -133,6 +134,7 @@ public OzoneManagerStateMachine(OzoneManagerRatisServer ratisServer,
133134
this.installSnapshotExecutor =
134135
HadoopExecutors.newSingleThreadExecutor(installSnapshotThreadFactory);
135136
this.nettyMetrics = NettyMetrics.create();
137+
this.successfulRequestHandler = new OzoneManagerSuccessfulRequestHandler(ozoneManager);
136138
}
137139

138140
/**
@@ -413,13 +415,13 @@ public CompletableFuture<Message> applyTransaction(TransactionContext trx) {
413415
ozoneManagerDoubleBuffer.acquireUnFlushedTransactions(1);
414416

415417
return CompletableFuture.supplyAsync(() -> runCommand(request, termIndex), executorService)
416-
.thenApply(this::processResponse);
418+
.thenApply(resp -> processResponse(request, resp, termIndex));
417419
} catch (Exception e) {
418420
return completeExceptionally(e);
419421
}
420422
}
421423

422-
private Message processResponse(OMResponse omResponse) {
424+
private Message processResponse(OMRequest request, OMResponse omResponse, TermIndex termIndex) {
423425
if (!omResponse.getSuccess()) {
424426
// INTERNAL_ERROR or METADATA_ERROR are considered as critical errors.
425427
// In such cases, OM must be terminated instead of completing the future exceptionally,
@@ -429,6 +431,10 @@ private Message processResponse(OMResponse omResponse) {
429431
} else if (omResponse.getStatus() == METADATA_ERROR) {
430432
terminate(omResponse, OMException.ResultCodes.METADATA_ERROR);
431433
}
434+
} else {
435+
// The operation completed successfully - hand off the request
436+
// so we can perform some post-actions
437+
successfulRequestHandler.handle(termIndex.getIndex(), request);
432438
}
433439

434440
// For successful response and non-critical errors, convert the response.

‎hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ratis/OzoneManagerSuccessfulRequestHandler.java‎

Lines changed: 184 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,184 @@
1+
/**
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
* <p>
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
* <p>
12+
* Unless required by applicable law or agreed to in writing, software
13+
* distributed under the License is distributed on an "AS IS" BASIS,
14+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15+
* See the License for the specific language governing permissions and
16+
* limitations under the License.
17+
*/
18+
19+
package org.apache.hadoop.ozone.om.ratis;
20+
21+
import java.io.IOException;
22+
import org.apache.commons.lang3.StringUtils;
23+
import org.apache.hadoop.hdds.utils.db.BatchOperation;
24+
import org.apache.hadoop.ozone.om.OzoneManager;
25+
import org.apache.hadoop.ozone.om.OMMetadataManager;
26+
import org.apache.hadoop.ozone.om.helpers.OmCompletedRequestInfo;
27+
import org.apache.hadoop.ozone.om.helpers.OmCompletedRequestInfo.OperationArgs;
28+
import org.apache.hadoop.ozone.om.helpers.OmCompletedRequestInfo.OperationType;
29+
import org.apache.hadoop.ozone.om.helpers.OmKeyArgs;
30+
import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos;
31+
import org.slf4j.Logger;
32+
import org.slf4j.LoggerFactory;
33+
34+
/**
35+
* This class is a simple hook on a successful write operation. It's
36+
* only purpose at the moment is to write an OmCompletedRequestInfo record to the DB
37+
*/
38+
public final class OzoneManagerSuccessfulRequestHandler {
39+
40+
private static final Logger LOG =
41+
LoggerFactory.getLogger(OzoneManagerSuccessfulRequestHandler.class);
42+
43+
private final OzoneManager ozoneManager;
44+
private final OMMetadataManager omMetadataManager;
45+
46+
public OzoneManagerSuccessfulRequestHandler(OzoneManager ozoneManager) {
47+
this.ozoneManager = ozoneManager;
48+
this.omMetadataManager = ozoneManager.getMetadataManager();
49+
}
50+
51+
public void handle(long trxLogIndex, OzoneManagerProtocolProtos.OMRequest omRequest) {
52+
53+
switch (omRequest.getCmdType()) {
54+
case CreateKey:
55+
logRequest("CreateKey", omRequest);
56+
storeCompletedRequestInfo(buildOmCompletedRequestInfo(trxLogIndex,
57+
omRequest.getCreateKeyRequest().getKeyArgs(),
58+
new OperationArgs.CreateKeyArgs()));
59+
break;
60+
case RenameKey:
61+
logRequest("RenameKey", omRequest);
62+
OzoneManagerProtocolProtos.RenameKeyRequest renameReq
63+
= (OzoneManagerProtocolProtos.RenameKeyRequest) omRequest.getRenameKeyRequest();
64+
65+
storeCompletedRequestInfo(buildOmCompletedRequestInfo(trxLogIndex,
66+
omRequest.getRenameKeyRequest().getKeyArgs(),
67+
new OperationArgs.RenameKeyArgs(renameReq.getToKeyName())));
68+
69+
break;
70+
case DeleteKey:
71+
logRequest("DeleteKey", omRequest);
72+
storeCompletedRequestInfo(buildOmCompletedRequestInfo(trxLogIndex,
73+
omRequest.getDeleteKeyRequest().getKeyArgs(),
74+
new OperationArgs.DeleteKeyArgs()));
75+
break;
76+
case CommitKey:
77+
logRequest("CommitKey", omRequest);
78+
storeCompletedRequestInfo(buildOmCompletedRequestInfo(trxLogIndex,
79+
omRequest.getCommitKeyRequest().getKeyArgs(),
80+
new OperationArgs.CommitKeyArgs()));
81+
break;
82+
case CreateDirectory:
83+
logRequest("CreateDirectory", omRequest);
84+
storeCompletedRequestInfo(buildOmCompletedRequestInfo(trxLogIndex,
85+
omRequest.getCreateDirectoryRequest().getKeyArgs(),
86+
new OperationArgs.CreateDirectoryArgs()));
87+
break;
88+
case CreateFile:
89+
logRequest("CreateFile", omRequest);
90+
91+
OzoneManagerProtocolProtos.CreateFileRequest createFileReq
92+
= (OzoneManagerProtocolProtos.CreateFileRequest) omRequest.getCreateFileRequest();
93+
94+
storeCompletedRequestInfo(buildOmCompletedRequestInfo(trxLogIndex,
95+
omRequest.getCreateFileRequest().getKeyArgs(),
96+
new OperationArgs.CreateFileArgs(createFileReq.getIsRecursive(),
97+
createFileReq.getIsOverwrite())));
98+
break;
99+
default:
100+
LOG.error("Unhandled cmdType={}", omRequest.getCmdType());
101+
break;
102+
}
103+
}
104+
105+
private static void logRequest(String label, OzoneManagerProtocolProtos.OMRequest omRequest) {
106+
if (LOG.isDebugEnabled()) {
107+
LOG.debug("---> {} {}", label, omRequest);
108+
}
109+
}
110+
111+
private OmCompletedRequestInfo buildOmCompletedRequestInfo(long trxLogIndex,
112+
OzoneManagerProtocolProtos.KeyArgs keyArgs,
113+
OperationArgs opArgs) {
114+
return OmCompletedRequestInfo.newBuilder()
115+
.setTrxLogIndex(trxLogIndex)
116+
.setVolumeName(keyArgs.getVolumeName())
117+
.setBucketName(keyArgs.getBucketName())
118+
.setKeyName(keyArgs.getKeyName())
119+
.setCreationTime(System.currentTimeMillis())
120+
.setOpArgs(opArgs)
121+
.build();
122+
}
123+
124+
private void storeCompletedRequestInfo(OmCompletedRequestInfo requestInfo) {
125+
if (LOG.isDebugEnabled()) {
126+
LOG.debug("Storing request info {}", requestInfo);
127+
}
128+
129+
// XXX: not sure if this string key is necessary. I added it as an
130+
// identifier which consumers of the ledger could use an efficient
131+
// (lexiographically sortable) key which could serve as a "seek
132+
// position" to continue reading where they left off (and to
133+
// persist to remember where they needed to carry on from). But
134+
// that may be unnecessary. TODO: can we just use a plain integer
135+
// key?
136+
String key = requestInfo.getDbKey();
137+
138+
// XXX: should this be part of an atomic db txn that happens at the end
139+
// of each replayed event or batch of events so that the completed
140+
// request info "ledger" table is consistent with the processed
141+
// raits events? e.g. OzoneManagerDoubleBuffer?
142+
143+
try (BatchOperation batchOperation = omMetadataManager.getStore()
144+
.initBatchOperation()) {
145+
146+
omMetadataManager.getCompletedRequestInfoTable().putWithBatch(batchOperation, key, requestInfo);
147+
148+
// TODO: cap the size of the table to some configured limit.
149+
//
150+
// The following code is taken from
151+
// https://github.com/apache/ozone/pull/8779/files#r2510853726
152+
//
153+
// ... as a suggested approach but I think it will need amended to
154+
// work here because the code seems to be predicated on all txnids
155+
// being held (and therefore we can count the nuber of IDs to
156+
// delete by subtracting new txnid from the first) whereas
157+
// CompletedRequestInfoTable only holds the details of a subset of
158+
// "interesting" write requests and therefore there are gaps in
159+
// the IDs.
160+
//
161+
// I'm not sure how best to approach this. A couple of( strawman)
162+
// ideas:
163+
//
164+
// 1. we could store every operation wherther interesting or not (or we
165+
// add dummy rows for the "non interesting" requests).
166+
// 2. we store an in memory count of the CompletedRequestInfoTable
167+
// which is initialized on startup and updated as rows are
168+
// added/cycled out. Therefore we know how many to cap the table
169+
// size as.
170+
//
171+
// TODO: revisit this
172+
//
173+
//omMetadataManager.getCompletedRequestInfoTable().deleteRangeWithBatch(batchOperation, 0L,
174+
// Math.max(lastTransaction.getIndex() - maxFlushedTransactionGap, 0L));
175+
176+
omMetadataManager.getStore().commitBatchOperation(batchOperation);
177+
178+
//} catch (IOException ex) {
179+
// LOG.error("Unable to write operation {}", requestInfo, ex);
180+
} catch (Exception ex) {
181+
LOG.error("Unable to write operation {}", requestInfo, ex);
182+
}
183+
}
184+
}

‎hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/ratis/TestOzoneManagerSuccessfulRequestHandler.java‎

Lines changed: 167 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,167 @@
1+
/**
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with this
4+
* work for additional information regarding copyright ownership. The ASF
5+
* licenses this file to you under the Apache License, Version 2.0 (the
6+
* "License"); you may not use this file except in compliance with the License.
7+
* You may obtain a copy of the License at
8+
* <p>
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
* <p>
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,WITHOUT
13+
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
14+
* License for the specific language governing permissions and limitations under
15+
* the License.
16+
*/
17+
package org.apache.hadoop.ozone.om.ratis;
18+
19+
import java.io.IOException;
20+
import java.util.UUID;
21+
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
22+
import org.apache.hadoop.hdds.utils.db.BatchOperation;
23+
import org.apache.hadoop.hdds.utils.db.DBStore;
24+
import org.apache.hadoop.hdds.utils.db.Table;
25+
import org.apache.hadoop.ozone.om.OMMetadataManager;
26+
import org.apache.hadoop.ozone.om.OzoneManager;
27+
import org.apache.hadoop.ozone.om.helpers.OmCompletedRequestInfo;
28+
import org.apache.hadoop.ozone.om.helpers.OmCompletedRequestInfo.OperationArgs;
29+
import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.CreateKeyRequest;
30+
import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.RenameKeyRequest;
31+
import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.KeyArgs;
32+
import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMRequest;
33+
import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.Type;
34+
import org.junit.jupiter.api.Assertions;
35+
import org.junit.jupiter.api.Test;
36+
import org.junit.jupiter.api.extension.ExtendWith;
37+
import org.junit.jupiter.api.io.TempDir;
38+
import org.mockito.Mock;
39+
import org.mockito.Mockito;
40+
import org.mockito.MockedConstruction;
41+
import org.mockito.ArgumentCaptor;
42+
import org.mockito.junit.jupiter.MockitoExtension;
43+
44+
import static org.assertj.core.api.Assertions.assertThat;
45+
import static org.mockito.Mockito.mock;
46+
import static org.mockito.Mockito.mockConstruction;
47+
import static org.mockito.Mockito.times;
48+
import static org.mockito.Mockito.when;
49+
import static org.mockito.Mockito.verify;
50+
51+
/**
52+
* Testing OzoneManagerSuccessfulRequestHandler class.
53+
*/
54+
@ExtendWith(MockitoExtension.class)
55+
public class TestOzoneManagerSuccessfulRequestHandler {
56+
57+
private static final String TEST_VOLUME_NAME = "testVol";
58+
private static final String TEST_BUCKET_NAME = "testBucket";
59+
private static final String TEST_KEY = "/foo/bar/baz/key";
60+
private static final String TEST_KEY_RENAMED = TEST_KEY + "_RENAMED";
61+
62+
private static final KeyArgs TEST_KEY_ARGS = KeyArgs.newBuilder()
63+
.setKeyName(TEST_KEY)
64+
.setVolumeName(TEST_VOLUME_NAME)
65+
.setBucketName(TEST_BUCKET_NAME)
66+
.build();
67+
68+
@Mock
69+
private OzoneManager ozoneManager;
70+
71+
@Mock
72+
private OMMetadataManager omMetadataManager;
73+
74+
@Mock
75+
private OzoneConfiguration configuration;
76+
77+
@Mock
78+
private DBStore dbStore;
79+
80+
@Mock
81+
private BatchOperation batchOperation;
82+
83+
@Mock
84+
private Table<String, OmCompletedRequestInfo> completedRequestInfoTable;
85+
86+
protected OMRequest createCreateKeyRequest() {
87+
CreateKeyRequest createKeyRequest = CreateKeyRequest.newBuilder()
88+
.setKeyArgs(TEST_KEY_ARGS).build();
89+
90+
return OMRequest.newBuilder()
91+
.setClientId(UUID.randomUUID().toString())
92+
.setCreateKeyRequest(createKeyRequest)
93+
.setCmdType(Type.CreateKey).build();
94+
}
95+
96+
protected OMRequest createRenameKeyRequest() {
97+
RenameKeyRequest renameKeyRequest = RenameKeyRequest.newBuilder()
98+
.setKeyArgs(TEST_KEY_ARGS).setToKeyName(TEST_KEY_RENAMED).build();
99+
100+
return OMRequest.newBuilder()
101+
.setClientId(UUID.randomUUID().toString())
102+
.setRenameKeyRequest(renameKeyRequest)
103+
.setCmdType(Type.RenameKey).build();
104+
}
105+
106+
@Test
107+
public void testCreateKeyRequest() throws IOException {
108+
109+
when(ozoneManager.getMetadataManager()).thenReturn(omMetadataManager);
110+
when(omMetadataManager.getCompletedRequestInfoTable()).thenReturn(completedRequestInfoTable);
111+
when(omMetadataManager.getStore()).thenReturn(dbStore);
112+
when(dbStore.initBatchOperation()).thenReturn(batchOperation);
113+
114+
OzoneManagerSuccessfulRequestHandler requestHandler
115+
= new OzoneManagerSuccessfulRequestHandler(ozoneManager);
116+
117+
requestHandler.handle(123L, createCreateKeyRequest());
118+
119+
ArgumentCaptor<BatchOperation> arg1 = ArgumentCaptor.forClass(BatchOperation.class);
120+
ArgumentCaptor<String> arg2 = ArgumentCaptor.forClass(String.class);
121+
ArgumentCaptor<OmCompletedRequestInfo> arg3 = ArgumentCaptor.forClass(OmCompletedRequestInfo.class);
122+
123+
verify(completedRequestInfoTable, times(1)).putWithBatch(arg1.capture(), arg2.capture(), arg3.capture());
124+
assertThat(arg1.getValue()).isEqualTo(batchOperation);
125+
126+
String key = arg2.getValue();
127+
assertThat(key).isEqualTo("00000000000000000123");
128+
129+
OmCompletedRequestInfo requestInfo = arg3.getValue();
130+
assertThat(requestInfo.getVolumeName()).isEqualTo(TEST_VOLUME_NAME);
131+
assertThat(requestInfo.getBucketName()).isEqualTo(TEST_BUCKET_NAME);
132+
assertThat(requestInfo.getKeyName()).isEqualTo(TEST_KEY);
133+
assertThat(requestInfo.getOpArgs()).isInstanceOf(OperationArgs.CreateKeyArgs.class);
134+
}
135+
136+
@Test
137+
public void testRenameKeyRequest() throws IOException {
138+
139+
when(ozoneManager.getMetadataManager()).thenReturn(omMetadataManager);
140+
when(omMetadataManager.getCompletedRequestInfoTable()).thenReturn(completedRequestInfoTable);
141+
when(omMetadataManager.getStore()).thenReturn(dbStore);
142+
when(dbStore.initBatchOperation()).thenReturn(batchOperation);
143+
144+
OzoneManagerSuccessfulRequestHandler requestHandler
145+
= new OzoneManagerSuccessfulRequestHandler(ozoneManager);
146+
147+
requestHandler.handle(124L, createRenameKeyRequest());
148+
149+
ArgumentCaptor<BatchOperation> arg1 = ArgumentCaptor.forClass(BatchOperation.class);
150+
ArgumentCaptor<String> arg2 = ArgumentCaptor.forClass(String.class);
151+
ArgumentCaptor<OmCompletedRequestInfo> arg3 = ArgumentCaptor.forClass(OmCompletedRequestInfo.class);
152+
153+
verify(completedRequestInfoTable, times(1)).putWithBatch(arg1.capture(), arg2.capture(), arg3.capture());
154+
assertThat(arg1.getValue()).isEqualTo(batchOperation);
155+
156+
String key = arg2.getValue();
157+
assertThat(key).isEqualTo("00000000000000000124");
158+
159+
OmCompletedRequestInfo requestInfo = arg3.getValue();
160+
assertThat(requestInfo.getVolumeName()).isEqualTo(TEST_VOLUME_NAME);
161+
assertThat(requestInfo.getBucketName()).isEqualTo(TEST_BUCKET_NAME);
162+
assertThat(requestInfo.getKeyName()).isEqualTo(TEST_KEY);
163+
assertThat(requestInfo.getOpArgs()).isInstanceOf(OperationArgs.RenameKeyArgs.class);
164+
OperationArgs.RenameKeyArgs opArgs = (OperationArgs.RenameKeyArgs) requestInfo.getOpArgs();
165+
assertThat(opArgs.getToKeyName()).isEqualTo(TEST_KEY_RENAMED);
166+
}
167+
}

0 commit comments

Comments
 (0)