Skip to content

Commit c9ca8f7

Browse files
author
Colm Dougan
committed
HDDS-14004. EventNotification: Capture data to the completed operation ledger table
1 parent 97c9862 commit c9ca8f7

10 files changed

Lines changed: 322 additions & 13 deletions
Lines changed: 155 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,155 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package org.apache.hadoop.ozone.om.ratis;
19+
20+
import org.apache.hadoop.ozone.om.OzoneManager;
21+
import org.apache.hadoop.ozone.om.helpers.OmCompletedRequestInfo;
22+
import org.apache.hadoop.ozone.om.helpers.OmCompletedRequestInfo.OperationArgs;
23+
import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos;
24+
import org.slf4j.Logger;
25+
import org.slf4j.LoggerFactory;
26+
27+
/**
28+
* This class when given a request will determine if the request is one of
29+
* a subset of write requests which we want to save to the
30+
* OmCompletedRequestInfo ledger and if so it will return the
31+
* appropriate instance.
32+
*
33+
* NOTE: as per elsewhere - this class may not need to exist if we go
34+
* with the approach of the subset of response handlers which we want to
35+
* populate a OmCompletedRequestInfo ledger row can do so themselves
36+
* (e.g. in OMKeyCommitResponse::addToBatch)
37+
*/
38+
public final class OzoneManagerCompletedRequestInfoProvider {
39+
40+
private static final Logger LOG =
41+
LoggerFactory.getLogger(OzoneManagerCompletedRequestInfoProvider.class);
42+
43+
private final OzoneManager ozoneManager;
44+
45+
public OzoneManagerCompletedRequestInfoProvider(OzoneManager ozoneManager) {
46+
this.ozoneManager = ozoneManager;
47+
}
48+
49+
public OmCompletedRequestInfo get(long trxLogIndex, OzoneManagerProtocolProtos.OMRequest omRequest) {
50+
51+
switch (omRequest.getCmdType()) {
52+
case CreateVolume:
53+
logRequest("CreateVolume", omRequest);
54+
return requestInfoBuilder(trxLogIndex, omRequest.getCmdType(), null)
55+
.setVolumeName(omRequest.getCreateVolumeRequest().getVolumeInfo().getVolume())
56+
.setOpArgs(new OperationArgs.CreateVolumeArgs())
57+
.build();
58+
59+
case DeleteVolume:
60+
logRequest("DeleteVolume", omRequest);
61+
return requestInfoBuilder(trxLogIndex, omRequest.getCmdType(), null)
62+
.setVolumeName(omRequest.getDeleteVolumeRequest().getVolumeName())
63+
.setOpArgs(new OperationArgs.DeleteVolumeArgs())
64+
.build();
65+
66+
case CreateBucket:
67+
logRequest("CreateBucket", omRequest);
68+
return requestInfoBuilder(trxLogIndex, omRequest.getCmdType(), null)
69+
.setVolumeName(omRequest.getCreateBucketRequest().getBucketInfo().getVolumeName())
70+
.setBucketName(omRequest.getCreateBucketRequest().getBucketInfo().getBucketName())
71+
.setOpArgs(new OperationArgs.CreateBucketArgs())
72+
.build();
73+
74+
case DeleteBucket:
75+
logRequest("DeleteBucket", omRequest);
76+
return requestInfoBuilder(trxLogIndex, omRequest.getCmdType(), null)
77+
.setVolumeName(omRequest.getDeleteBucketRequest().getVolumeName())
78+
.setBucketName(omRequest.getDeleteBucketRequest().getBucketName())
79+
.setOpArgs(new OperationArgs.DeleteBucketArgs())
80+
.build();
81+
82+
case CreateKey:
83+
logRequest("CreateKey", omRequest);
84+
return requestInfoBuilder(trxLogIndex, omRequest.getCmdType(), omRequest.getCreateKeyRequest().getKeyArgs())
85+
.setOpArgs(new OperationArgs.CreateKeyArgs())
86+
.build();
87+
88+
case RenameKey:
89+
logRequest("RenameKey", omRequest);
90+
OzoneManagerProtocolProtos.RenameKeyRequest renameReq
91+
= (OzoneManagerProtocolProtos.RenameKeyRequest) omRequest.getRenameKeyRequest();
92+
93+
return requestInfoBuilder(trxLogIndex, omRequest.getCmdType(), omRequest.getRenameKeyRequest().getKeyArgs())
94+
.setOpArgs(new OperationArgs.RenameKeyArgs(renameReq.getToKeyName()))
95+
.build();
96+
97+
case DeleteKey:
98+
logRequest("DeleteKey", omRequest);
99+
return requestInfoBuilder(trxLogIndex, omRequest.getCmdType(), omRequest.getDeleteKeyRequest().getKeyArgs())
100+
.setOpArgs(new OperationArgs.DeleteKeyArgs())
101+
.build();
102+
103+
case CommitKey:
104+
logRequest("CommitKey", omRequest);
105+
return requestInfoBuilder(trxLogIndex, omRequest.getCmdType(), omRequest.getCommitKeyRequest().getKeyArgs())
106+
.setOpArgs(new OperationArgs.CommitKeyArgs())
107+
.build();
108+
109+
case CreateDirectory:
110+
logRequest("CreateDirectory", omRequest);
111+
return requestInfoBuilder(trxLogIndex, omRequest.getCmdType(), omRequest.getCreateDirectoryRequest().getKeyArgs())
112+
.setOpArgs(new OperationArgs.CreateDirectoryArgs())
113+
.build();
114+
115+
case CreateFile:
116+
logRequest("CreateFile", omRequest);
117+
118+
OzoneManagerProtocolProtos.CreateFileRequest createFileReq
119+
= (OzoneManagerProtocolProtos.CreateFileRequest) omRequest.getCreateFileRequest();
120+
121+
return requestInfoBuilder(trxLogIndex, omRequest.getCmdType(), omRequest.getCreateFileRequest().getKeyArgs())
122+
.setOpArgs(new OperationArgs.CreateFileArgs(createFileReq.getIsRecursive(),
123+
createFileReq.getIsOverwrite()))
124+
.build();
125+
126+
default:
127+
LOG.debug("Unhandled cmdType={}", omRequest.getCmdType());
128+
return null;
129+
}
130+
}
131+
132+
private static void logRequest(String label, OzoneManagerProtocolProtos.OMRequest omRequest) {
133+
if (LOG.isDebugEnabled()) {
134+
LOG.debug("---> {} {}", label, omRequest);
135+
}
136+
}
137+
138+
private OmCompletedRequestInfo.Builder requestInfoBuilder(long trxLogIndex,
139+
OzoneManagerProtocolProtos.Type cmdType,
140+
OzoneManagerProtocolProtos.KeyArgs keyArgs) {
141+
142+
OmCompletedRequestInfo.Builder builder = OmCompletedRequestInfo.newBuilder()
143+
.setTrxLogIndex(trxLogIndex)
144+
.setCmdType(cmdType)
145+
.setCreationTime(System.currentTimeMillis());
146+
147+
if (keyArgs != null) {
148+
builder.setVolumeName(keyArgs.getVolumeName());
149+
builder.setBucketName(keyArgs.getBucketName());
150+
builder.setKeyName(keyArgs.getKeyName());
151+
}
152+
153+
return builder;
154+
}
155+
}

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

Lines changed: 30 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -44,6 +44,7 @@
4444
import org.apache.hadoop.ozone.om.OMMetadataManager;
4545
import org.apache.hadoop.ozone.om.S3SecretManager;
4646
import org.apache.hadoop.ozone.om.codec.OMDBDefinition;
47+
import org.apache.hadoop.ozone.om.helpers.OmCompletedRequestInfo;
4748
import org.apache.hadoop.ozone.om.response.CleanupTableInfo;
4849
import org.apache.hadoop.ozone.om.response.OMClientResponse;
4950
import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos;
@@ -106,10 +107,13 @@ public final class OzoneManagerDoubleBuffer {
106107
private static class Entry {
107108
private final TermIndex termIndex;
108109
private final OMClientResponse response;
110+
// strawman approach: see comments in addToBatch()
111+
private final OmCompletedRequestInfo completedRequestInfo;
109112

110-
Entry(TermIndex termIndex, OMClientResponse response) {
113+
Entry(TermIndex termIndex, OMClientResponse response, OmCompletedRequestInfo completedRequestInfo) {
111114
this.termIndex = termIndex;
112115
this.response = response;
116+
this.completedRequestInfo = completedRequestInfo;
113117
}
114118

115119
TermIndex getTermIndex() {
@@ -119,6 +123,10 @@ TermIndex getTermIndex() {
119123
OMClientResponse getResponse() {
120124
return response;
121125
}
126+
127+
OmCompletedRequestInfo getCompletedRequestInfo() {
128+
return completedRequestInfo;
129+
}
122130
}
123131

124132
/**
@@ -404,11 +412,29 @@ private String addToBatch(Queue<Entry> buffer, BatchOperation batchOperation) {
404412
for (Entry entry: buffer) {
405413
OMClientResponse response = entry.getResponse();
406414
OMResponse omResponse = response.getOMResponse();
415+
OmCompletedRequestInfo completedRequestInfo = entry.getCompletedRequestInfo();
407416
lastTraceId = omResponse.getTraceID();
408417

409418
try {
410419
addToBatchWithTrace(omResponse,
411420
() -> response.checkAndUpdateDB(omMetadataManager, batchOperation));
421+
422+
// This is a strawman approach and requires some discussion
423+
// with the community on approach.
424+
//
425+
// TODO: would it be better to have each type of request we want
426+
// to capture populate a row in the ledger in the addToBatch
427+
// callback of the response (triggered above) e.g.
428+
// OMKeyCommitResponse::addToBatch could generate the suitable
429+
// OmCompletedRequestInfo row and populate (Q: if we went with
430+
// that approach then would it have access to all the necessary
431+
// request parameters? e.g. termIndex.getIndex(), cmdType and
432+
// all captured request parameters
433+
if (completedRequestInfo != null) {
434+
omMetadataManager.getCompletedRequestInfoTable().putWithBatch(
435+
batchOperation, completedRequestInfo.getTrxLogIndex(), completedRequestInfo);
436+
}
437+
412438
} catch (IOException ex) {
413439
// During Adding to RocksDB batch entry got an exception.
414440
// We should terminate the OM.
@@ -554,8 +580,9 @@ private void terminate(Throwable t, int status, OMResponse omResponse) {
554580
/**
555581
* Add OmResponseBufferEntry to buffer.
556582
*/
557-
public synchronized void add(OMClientResponse response, TermIndex termIndex) {
558-
currentBuffer.add(new Entry(termIndex, response));
583+
public synchronized void add(OMClientResponse response, TermIndex termIndex,
584+
OmCompletedRequestInfo completedRequestInfo) {
585+
currentBuffer.add(new Entry(termIndex, response, completedRequestInfo));
559586
notify();
560587
}
561588

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -629,7 +629,7 @@ private OMResponse createErrorResponse(
629629
}
630630
OMResponse omResponse = omResponseBuilder.build();
631631
OMClientResponse omClientResponse = new DummyOMClientResponse(omResponse);
632-
ozoneManagerDoubleBuffer.add(omClientResponse, termIndex);
632+
ozoneManagerDoubleBuffer.add(omClientResponse, termIndex, null);
633633
return omResponse;
634634
}
635635

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/upgrade/OMPrepareRequest.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -103,7 +103,7 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, Execut
103103
// the snapshot index in the prepared state.
104104
OzoneManagerDoubleBuffer doubleBuffer =
105105
ozoneManager.getOmRatisServer().getOmStateMachine().getOzoneManagerDoubleBuffer();
106-
doubleBuffer.add(response, context.getTermIndex());
106+
doubleBuffer.add(response, context.getTermIndex(), null);
107107

108108
OzoneManagerRatisServer omRatisServer = ozoneManager.getOmRatisServer();
109109
final RaftServer.Division division = omRatisServer.getServerDivision();

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/protocolPB/OzoneManagerRequestHandler.java

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -75,6 +75,7 @@
7575
import org.apache.hadoop.ozone.om.helpers.ListOpenFilesResult;
7676
import org.apache.hadoop.ozone.om.helpers.OMAuditLogger;
7777
import org.apache.hadoop.ozone.om.helpers.OmBucketInfo;
78+
import org.apache.hadoop.ozone.om.helpers.OmCompletedRequestInfo;
7879
import org.apache.hadoop.ozone.om.helpers.OmKeyArgs;
7980
import org.apache.hadoop.ozone.om.helpers.OmKeyInfo;
8081
import org.apache.hadoop.ozone.om.helpers.OmMultipartUploadList;
@@ -91,6 +92,7 @@
9192
import org.apache.hadoop.ozone.om.helpers.TenantStateList;
9293
import org.apache.hadoop.ozone.om.helpers.TenantUserInfoValue;
9394
import org.apache.hadoop.ozone.om.helpers.TenantUserList;
95+
import org.apache.hadoop.ozone.om.ratis.OzoneManagerCompletedRequestInfoProvider;
9496
import org.apache.hadoop.ozone.om.ratis.utils.OzoneManagerRatisUtils;
9597
import org.apache.hadoop.ozone.om.request.OMClientRequest;
9698
import org.apache.hadoop.ozone.om.request.util.OmResponseUtil;
@@ -166,6 +168,7 @@
166168
import org.apache.hadoop.ozone.upgrade.UpgradeFinalization.StatusAndMessages;
167169
import org.apache.hadoop.ozone.util.PayloadUtils;
168170
import org.apache.hadoop.ozone.util.ProtobufUtils;
171+
import org.apache.ratis.server.protocol.TermIndex;
169172
import org.slf4j.Logger;
170173
import org.slf4j.LoggerFactory;
171174

@@ -177,10 +180,12 @@ public class OzoneManagerRequestHandler implements RequestHandler {
177180
static final Logger LOG =
178181
LoggerFactory.getLogger(OzoneManagerRequestHandler.class);
179182
private final OzoneManager impl;
183+
private final OzoneManagerCompletedRequestInfoProvider completedRequestInfoProvider;
180184
private FaultInjector injector;
181185

182186
public OzoneManagerRequestHandler(OzoneManager om) {
183187
this.impl = om;
188+
this.completedRequestInfoProvider = new OzoneManagerCompletedRequestInfoProvider(om);
184189
}
185190

186191
//TODO simplify it to make it shorter
@@ -426,6 +431,12 @@ public OMClientResponse handleWriteRequestImpl(OMRequest omRequest, ExecutionCon
426431
}
427432
}
428433

434+
@Override
435+
public OmCompletedRequestInfo handleSuccessfulWriteRequestImpl(TermIndex termIndex, OMRequest omRequest)
436+
throws IOException {
437+
return completedRequestInfoProvider.get(termIndex.getIndex(), omRequest);
438+
}
439+
429440
@VisibleForTesting
430441
public void setInjector(FaultInjector injector) {
431442
this.injector = injector;

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/protocolPB/RequestHandler.java

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -20,11 +20,13 @@
2020
import java.io.IOException;
2121
import org.apache.hadoop.ozone.om.exceptions.OMException;
2222
import org.apache.hadoop.ozone.om.execution.flowcontrol.ExecutionContext;
23+
import org.apache.hadoop.ozone.om.helpers.OmCompletedRequestInfo;
2324
import org.apache.hadoop.ozone.om.ratis.OzoneManagerDoubleBuffer;
2425
import org.apache.hadoop.ozone.om.response.OMClientResponse;
2526
import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMRequest;
2627
import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMResponse;
2728
import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.Type;
29+
import org.apache.ratis.server.protocol.TermIndex;
2830

2931
/**
3032
* Handler to handleRequest the OmRequests.
@@ -61,7 +63,11 @@ default OMClientResponse handleWriteRequest(OMRequest omRequest, ExecutionContex
6163
OzoneManagerDoubleBuffer ozoneManagerDoubleBuffer) throws IOException {
6264
final OMClientResponse response = handleWriteRequestImpl(omRequest, context);
6365
if (omRequest.getCmdType() != Type.Prepare) {
64-
ozoneManagerDoubleBuffer.add(response, context.getTermIndex());
66+
67+
final OmCompletedRequestInfo omCompletedRequestInfo = handleSuccessfulWriteRequestImpl(
68+
context.getTermIndex(), omRequest);
69+
70+
ozoneManagerDoubleBuffer.add(response, context.getTermIndex(), omCompletedRequestInfo);
6571
}
6672
return response;
6773
}
@@ -74,4 +80,6 @@ default OMClientResponse handleWriteRequest(OMRequest omRequest, ExecutionContex
7480
* @return OMClientResponse
7581
*/
7682
OMClientResponse handleWriteRequestImpl(OMRequest omRequest, ExecutionContext context) throws IOException;
83+
84+
OmCompletedRequestInfo handleSuccessfulWriteRequestImpl(TermIndex index, OMRequest omRequest) throws IOException;
7785
}

0 commit comments

Comments
 (0)