Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -66,10 +66,10 @@ public abstract class AbstractBolt extends BaseRichBolt {
@Getter(AccessLevel.PROTECTED)
private transient Integer taskId;

@Getter(AccessLevel.PROTECTED)
@Getter(AccessLevel.PUBLIC)

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Just out of curiosity, you decided to point out explicitly AccessLevel.PUBLIC only to keep the common behaviour here, like as it is in the cases for AccessLevel.PROTECTED?

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It is a part of an interface now, so it has to be public

private transient Tuple currentTuple;

@Getter(AccessLevel.PROTECTED)
@Getter(AccessLevel.PUBLIC)
@Setter(AccessLevel.PROTECTED)
private transient CommandContext commandContext;

Expand Down Expand Up @@ -137,7 +137,14 @@ protected void emit(String stream, List<Object> payload) {
output.emit(stream, payload);
}

protected void emitWithContext(String stream, Tuple input, Values payload) {
/**
* Emitting (sending a message) the given payload with the anchor based on the tuple to the given stream.
* This guarantees that the message will be eventually processed by Storm even in cases of the target bolt failure.
* @param stream Storm stream to send the message to
* @param input current tuple; will be used as an anchor
* @param payload the message payload
*/
public void emitWithContext(String stream, Tuple input, Values payload) {
payload.add(getCommandContext());
log.debug("emit tuple into {} stream: {}", stream, payload);
getOutput().emit(stream, input, payload);
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,29 @@
/* Copyright 2023 Telstra Open Source
*
* Licensed 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.openkilda.wfm;

import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

/**
* This interface represents something that has CommandContext and
* knows how to emit the payload via the given Storm stream.
*/
public interface ContextEmittable {
CommandContext getCommandContext();

void emitWithContext(String stream, Tuple input, Values payload);
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,40 @@
/* Copyright 2023 Telstra Open Source
*
* Licensed 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.openkilda.wfm;

import org.openkilda.messaging.info.InfoMessage;
import org.openkilda.wfm.share.history.model.FlowHistoryHolder;

import lombok.NonNull;
import org.apache.storm.tuple.Values;

/**
* This is a default implementation for sending history updates that are accessible later via HistoryService.
* A topology that uses the bolt implementing this interface must register a path to history bolt.
* That is, in a topology that extends AbstractTopology, declare a Kafka bolt and specify a component ID
* that is a source of history updates in the fields grouping and declare a stream.
* TODO add a readme.md with the details and example.
*/
public interface DefaultHistoryUpdateCarrier extends HistoryUpdateCarrier, TupleIdentifiable, ContextEmittable {

@Override
default void sendHistoryUpdate(@NonNull FlowHistoryHolder historyHolder) {
InfoMessage message = new InfoMessage(historyHolder, getCommandContext().getCreateTime(),
getCommandContext().getCorrelationId());
emitWithContext(getHistoryStreamName(), getCurrentTuple(),
new Values(historyHolder.getTaskId(), message));
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,7 @@
* limitations under the License.
*/

package org.openkilda.wfm.topology.flowhs.service.common;
package org.openkilda.wfm;

import org.openkilda.wfm.share.history.model.FlowHistoryHolder;

Expand All @@ -25,4 +25,8 @@ public interface HistoryUpdateCarrier {
* Sends main events to history bolt.
*/
void sendHistoryUpdate(FlowHistoryHolder historyHolder);

default String getHistoryStreamName() {
return "HUB_TO_HISTORY_TOPOLOGY_SENDER";
}
Comment on lines +29 to +31

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Would it be more convenient to keep HUB_TO_HISTORY_TOPOLOGY_SENDER constant in the Stream enum?

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, I have a thought that it needs to be in some place that is accessible in the bolts and the service at the same time. We need this constant when configuring topologies, so it would be convenient to have some unified approach of adding. If it is a helper method then it can't be enum that we have in the topology package.
This is a task to think about. But once it is done, we can change this method without changing the rest of the code.

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think getHistoryStreamName() must be used in declareOutputFields

}
Original file line number Diff line number Diff line change
@@ -0,0 +1,25 @@
/* Copyright 2023 Telstra Open Source
*
* Licensed 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.openkilda.wfm;

import org.apache.storm.tuple.Tuple;

/**
* This interface represents something that has a distinct tuple that can be used as an anchor in Storm.
*/
public interface TupleIdentifiable {
Tuple getCurrentTuple();
}
Original file line number Diff line number Diff line change
Expand Up @@ -41,10 +41,10 @@
import org.openkilda.pce.PathComputerFactory;
import org.openkilda.persistence.PersistenceManager;
import org.openkilda.server42.control.messaging.flowrtt.ActivateFlowMonitoringInfoData;
import org.openkilda.wfm.DefaultHistoryUpdateCarrier;
import org.openkilda.wfm.error.PipelineException;
import org.openkilda.wfm.share.flow.resources.FlowResourcesConfig;
import org.openkilda.wfm.share.flow.resources.FlowResourcesManager;
import org.openkilda.wfm.share.history.model.FlowHistoryHolder;
import org.openkilda.wfm.share.hubandspoke.HubBolt;
import org.openkilda.wfm.share.utils.KeyProvider;
import org.openkilda.wfm.share.zk.ZkStreams;
Expand All @@ -66,7 +66,7 @@
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

public class FlowCreateHubBolt extends HubBolt implements FlowGenericCarrier {
public class FlowCreateHubBolt extends HubBolt implements FlowGenericCarrier, DefaultHistoryUpdateCarrier {
private final FlowCreateConfig config;
private final PathComputerConfig pathComputerConfig;
private final FlowResourcesConfig flowResourcesConfig;
Expand Down Expand Up @@ -161,14 +161,6 @@ public void sendNorthboundResponse(@NonNull Message message) {
emitWithContext(Stream.HUB_TO_NB_RESPONSE_SENDER.name(), getCurrentTuple(), new Values(currentKey, message));
}

@Override
public void sendHistoryUpdate(@NonNull FlowHistoryHolder historyHolder) {
InfoMessage message = new InfoMessage(historyHolder, getCommandContext().getCreateTime(),
getCommandContext().getCorrelationId());
emitWithContext(Stream.HUB_TO_HISTORY_TOPOLOGY_SENDER.name(), getCurrentTuple(),
new Values(historyHolder.getTaskId(), message));
}

@Override
public void cancelTimeoutCallback(@NonNull String key) {
cancelCallback(key);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -26,10 +26,10 @@
import org.openkilda.model.PathId;
import org.openkilda.model.SwitchId;
import org.openkilda.wfm.CommandContext;
import org.openkilda.wfm.HistoryUpdateCarrier;
import org.openkilda.wfm.share.flow.resources.FlowResources;
import org.openkilda.wfm.topology.flowhs.model.RequestedFlow;
import org.openkilda.wfm.topology.flowhs.service.FlowProcessingEventListener;
import org.openkilda.wfm.topology.flowhs.service.common.HistoryUpdateCarrier;
import org.openkilda.wfm.topology.flowhs.service.common.NorthboundResponseCarrier;

import lombok.Getter;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -18,13 +18,13 @@
import static java.util.Collections.emptyList;

import org.openkilda.wfm.CommandContext;
import org.openkilda.wfm.HistoryUpdateCarrier;
import org.openkilda.wfm.share.history.model.FlowDumpData;
import org.openkilda.wfm.share.history.model.FlowEventData;
import org.openkilda.wfm.share.history.model.FlowHistoryData;
import org.openkilda.wfm.share.history.model.FlowHistoryHolder;
import org.openkilda.wfm.share.utils.KeyProvider;
import org.openkilda.wfm.topology.flowhs.service.common.FlowHistoryCarrier;
import org.openkilda.wfm.topology.flowhs.service.common.HistoryUpdateCarrier;
import org.openkilda.wfm.topology.flowhs.service.common.NorthboundResponseCarrier;
import org.openkilda.wfm.topology.flowhs.service.common.ProcessingEventListener;

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -21,8 +21,8 @@
import org.openkilda.floodlight.api.response.rulemanager.SpeakerCommandResponse;
import org.openkilda.model.SwitchId;
import org.openkilda.wfm.CommandContext;
import org.openkilda.wfm.HistoryUpdateCarrier;
import org.openkilda.wfm.topology.flowhs.service.FlowProcessingEventListener;
import org.openkilda.wfm.topology.flowhs.service.common.HistoryUpdateCarrier;
import org.openkilda.wfm.topology.flowhs.service.common.NorthboundResponseCarrier;

import lombok.Getter;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -31,12 +31,12 @@
import org.openkilda.pce.GetHaPathsResult;
import org.openkilda.pce.HaPath;
import org.openkilda.wfm.CommandContext;
import org.openkilda.wfm.HistoryUpdateCarrier;
import org.openkilda.wfm.share.flow.resources.HaFlowResources;
import org.openkilda.wfm.share.flow.resources.HaPathIdsPair;
import org.openkilda.wfm.share.flow.resources.HaPathIdsPair.HaFlowPathIds;
import org.openkilda.wfm.topology.flowhs.fsm.common.context.SpeakerResponseContext;
import org.openkilda.wfm.topology.flowhs.service.FlowProcessingEventListener;
import org.openkilda.wfm.topology.flowhs.service.common.HistoryUpdateCarrier;
import org.openkilda.wfm.topology.flowhs.service.common.NorthboundResponseCarrier;

import lombok.Getter;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -20,10 +20,10 @@
import org.openkilda.model.PathId;
import org.openkilda.persistence.PersistenceManager;
import org.openkilda.persistence.repositories.HaFlowPathRepository;
import org.openkilda.wfm.HistoryUpdateCarrier;
import org.openkilda.wfm.share.flow.resources.HaPathIdsPair;
import org.openkilda.wfm.topology.flowhs.fsm.common.FlowProcessingWithHistorySupportFsm;
import org.openkilda.wfm.topology.flowhs.fsm.common.actions.BaseFlowPathRemovalAction;
import org.openkilda.wfm.topology.flowhs.service.common.HistoryUpdateCarrier;
import org.openkilda.wfm.topology.flowhs.service.haflow.history.HaFlowHistory;
import org.openkilda.wfm.topology.flowhs.service.haflow.history.HaFlowHistoryService;

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -19,11 +19,11 @@

import org.openkilda.model.HaFlow;
import org.openkilda.persistence.PersistenceManager;
import org.openkilda.wfm.HistoryUpdateCarrier;
import org.openkilda.wfm.share.flow.resources.FlowResourcesManager;
import org.openkilda.wfm.share.flow.resources.HaFlowResources;
import org.openkilda.wfm.topology.flowhs.fsm.common.HaFlowPathSwappingFsm;
import org.openkilda.wfm.topology.flowhs.fsm.common.context.SpeakerResponseContext;
import org.openkilda.wfm.topology.flowhs.service.common.HistoryUpdateCarrier;
import org.openkilda.wfm.topology.flowhs.service.haflow.history.HaFlowHistory;
import org.openkilda.wfm.topology.flowhs.service.haflow.history.HaFlowHistoryService;

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -16,13 +16,13 @@
package org.openkilda.wfm.topology.flowhs.fsm.haflow.reroute.actions;

import org.openkilda.persistence.PersistenceManager;
import org.openkilda.wfm.HistoryUpdateCarrier;
import org.openkilda.wfm.share.metrics.TimedExecution;
import org.openkilda.wfm.topology.flowhs.fsm.common.actions.haflow.BaseHaFlowPathRemovalAction;
import org.openkilda.wfm.topology.flowhs.fsm.haflow.reroute.HaFlowRerouteContext;
import org.openkilda.wfm.topology.flowhs.fsm.haflow.reroute.HaFlowRerouteFsm;
import org.openkilda.wfm.topology.flowhs.fsm.haflow.reroute.HaFlowRerouteFsm.Event;
import org.openkilda.wfm.topology.flowhs.fsm.haflow.reroute.HaFlowRerouteFsm.State;
import org.openkilda.wfm.topology.flowhs.service.common.HistoryUpdateCarrier;

import lombok.extern.slf4j.Slf4j;

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -16,12 +16,12 @@
package org.openkilda.wfm.topology.flowhs.fsm.haflow.update.actions;

import org.openkilda.persistence.PersistenceManager;
import org.openkilda.wfm.HistoryUpdateCarrier;
import org.openkilda.wfm.topology.flowhs.fsm.common.actions.haflow.BaseHaFlowPathRemovalAction;
import org.openkilda.wfm.topology.flowhs.fsm.haflow.update.HaFlowUpdateContext;
import org.openkilda.wfm.topology.flowhs.fsm.haflow.update.HaFlowUpdateFsm;
import org.openkilda.wfm.topology.flowhs.fsm.haflow.update.HaFlowUpdateFsm.Event;
import org.openkilda.wfm.topology.flowhs.fsm.haflow.update.HaFlowUpdateFsm.State;
import org.openkilda.wfm.topology.flowhs.service.common.HistoryUpdateCarrier;

import lombok.extern.slf4j.Slf4j;

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -21,8 +21,8 @@
import org.openkilda.messaging.info.stats.StatsNotification;
import org.openkilda.messaging.info.stats.UpdateFlowPathInfo;
import org.openkilda.model.SwitchId;
import org.openkilda.wfm.HistoryUpdateCarrier;
import org.openkilda.wfm.topology.flowhs.model.RequestedFlow;
import org.openkilda.wfm.topology.flowhs.service.common.HistoryUpdateCarrier;
import org.openkilda.wfm.topology.flowhs.service.common.LifecycleEventCarrier;
import org.openkilda.wfm.topology.flowhs.service.common.NorthboundResponseCarrier;

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -15,10 +15,10 @@

package org.openkilda.wfm.topology.flowhs.service.haflow.history;

import org.openkilda.wfm.HistoryUpdateCarrier;
import org.openkilda.wfm.share.history.model.FlowHistoryHolder;
import org.openkilda.wfm.share.history.model.HaFlowEventData;
import org.openkilda.wfm.share.history.model.HaFlowHistoryData;
import org.openkilda.wfm.topology.flowhs.service.common.HistoryUpdateCarrier;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.ObjectUtils;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -23,12 +23,12 @@
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;

import org.openkilda.wfm.HistoryUpdateCarrier;
import org.openkilda.wfm.share.history.model.FlowHistoryHolder;
import org.openkilda.wfm.share.history.model.HaFlowDumpData;
import org.openkilda.wfm.share.history.model.HaFlowEventData;
import org.openkilda.wfm.share.history.model.HaFlowEventData.Event;
import org.openkilda.wfm.share.history.model.HaFlowEventData.Initiator;
import org.openkilda.wfm.topology.flowhs.service.common.HistoryUpdateCarrier;

import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
Expand Down