bejancsaba commented on a change in pull request #5755: URL: https://github.com/apache/nifi/pull/5755#discussion_r806855762
########## File path: minifi/minifi-c2/minifi-c2-service/src/main/java/org/apache/nifi/minifi/c2/service/SimpleC2ProtocolService.java ########## @@ -0,0 +1,120 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.nifi.minifi.c2.service; + +import org.apache.nifi.c2.protocol.api.C2Heartbeat; +import org.apache.nifi.c2.protocol.api.C2HeartbeatResponse; +import org.apache.nifi.c2.protocol.api.C2Operation; +import org.apache.nifi.c2.protocol.api.C2OperationAck; +import org.apache.nifi.c2.protocol.api.C2OperationState; +import org.apache.nifi.c2.protocol.api.OperandType; +import org.apache.nifi.c2.protocol.api.OperationState; +import org.apache.nifi.c2.protocol.api.OperationType; +import org.apache.nifi.minifi.c2.util.LogMarkerUtil; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.slf4j.Marker; +import org.springframework.stereotype.Service; + +import java.util.Collections; +import java.util.HashSet; +import java.util.List; +import java.util.Set; +import java.util.UUID; + +@Service +public class SimpleC2ProtocolService implements C2ProtocolService { + + private static final Logger logger = LoggerFactory.getLogger(SimpleC2ProtocolService.class); + + private static final Set<String> issuedOperationIds = new HashSet<>(); + + public SimpleC2ProtocolService() { + } + + @Override + public void processOperationAck(final C2OperationAck operationAck, final C2ProtocolContext context) { + // This service assumes there is a single Operation UPDATE to pass over the updated flow + Marker marker = LogMarkerUtil.getMarker(operationAck); + + logger.debug(marker, "Received operation acknowledgement: {}; {}", operationAck, context); + // Remove the operator ID from the list of issued operations and log the state + final String operationId = operationAck.getOperationId(); + try { + OperationState opState = OperationState.DONE; + String details = null; + + /* Partial applications are rare and only happen when an operation consists of updating multiple config + * items and some succeed ( we don't yet have the concept of rollback in agents ). + * Fully Applied yields an operation success. + * Operation Not Understood and Not Applied give little details but also will result in Operation Failure. + * We should explore if providing textual details. */ + final C2OperationState c2OperationState = operationAck.getOperationState(); + if (null != c2OperationState) { + details = c2OperationState.getDetails(); + if (c2OperationState.getState() != C2OperationState.OperationState.FULLY_APPLIED) { + opState = OperationState.FAILED; + } + } + + if (!issuedOperationIds.remove(operationId)) { + logger.warn(marker, "Operation with ID " + operationId + " has either already been acknowledged or is unknown to this server"); + } else if (null != c2OperationState) { + final C2OperationState.OperationState operationState = c2OperationState.getState(); + logger.debug("Operation with ID " + operationId + " acknowledged with a state of " + operationState.name() + "(" + opState.name() + "), details = " + + (details == null ? "" : details)); + } + + // Optionally, an acknowledgement can include some of the info normally passed in a heartbeat. + // If this info is present, process it as a heartbeat, so we update our latest known state of the agent. + if (operationAck.getAgentInfo() != null + || operationAck.getDeviceInfo() != null + || operationAck.getFlowInfo() != null) { + final C2Heartbeat heartbeatInfo = toHeartbeat(operationAck); + logger.trace(marker, "Operation acknowledgement contains additional info. Processing as heartbeat: {}", heartbeatInfo); + processHeartbeat(heartbeatInfo, context); + } + + } catch (final Exception e) { + logger.warn("Encountered exception while processing operation ack", e); + } + } + + @Override + public C2HeartbeatResponse processHeartbeat(final C2Heartbeat heartbeat, final C2ProtocolContext context) { Review comment: I think for now what we have should be good I was just worried that each time the unchanged flow is re-downloaded unnecessarily. If the agent doesn't apply the flow if it is unchanged we should be good and I agree we should make this more robust when the full C2 implementation comes I suppose it is not in scope for this change. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@nifi.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
