/* * Copyright (c) 2015 Cisco Systems, Inc. and others. All rights reserved. * * This program and the accompanying materials are made available under the * terms of the Eclipse Public License v1.0 which accompanies this distribution, * and is available at http://www.eclipse.org/legal/epl-v10.html */ package org.opendaylight.openflowplugin.impl.services.sal; import com.google.common.util.concurrent.FutureCallback; import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; import com.google.common.util.concurrent.MoreExecutors; import com.google.common.util.concurrent.SettableFuture; import java.util.ArrayList; import java.util.List; import org.opendaylight.openflowplugin.api.OFConstants; import org.opendaylight.openflowplugin.api.openflow.FlowGroupStatus; import org.opendaylight.openflowplugin.api.openflow.device.DeviceContext; import org.opendaylight.openflowplugin.api.openflow.device.RequestContextStack; import org.opendaylight.openflowplugin.api.openflow.registry.flow.DeviceFlowRegistry; import org.opendaylight.openflowplugin.api.openflow.registry.flow.FlowDescriptor; import org.opendaylight.openflowplugin.api.openflow.registry.flow.FlowRegistryKey; import org.opendaylight.openflowplugin.impl.registry.flow.FlowDescriptorFactory; import org.opendaylight.openflowplugin.impl.registry.flow.FlowRegistryKeyFactory; import org.opendaylight.openflowplugin.impl.services.multilayer.MultiLayerFlowService; import org.opendaylight.openflowplugin.impl.services.singlelayer.SingleLayerFlowService; import org.opendaylight.openflowplugin.impl.util.ErrorUtil; import org.opendaylight.openflowplugin.impl.util.FlowCreatorUtil; import org.opendaylight.openflowplugin.openflow.md.core.sal.convertor.ConvertorExecutor; import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowId; import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.Flow; import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.AddFlowInput; import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.AddFlowInputBuilder; import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.AddFlowOutput; import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.RemoveFlowInput; import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.RemoveFlowInputBuilder; import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.RemoveFlowOutput; import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.SalFlowService; import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.UpdateFlowInput; import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.UpdateFlowOutput; import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.flow.update.OriginalFlow; import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.flow.update.UpdatedFlow; import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.types.rev131026.FlowRef; import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.FlowModInputBuilder; import org.opendaylight.yangtools.yang.common.ErrorType; import org.opendaylight.yangtools.yang.common.RpcError; import org.opendaylight.yangtools.yang.common.RpcResult; import org.opendaylight.yangtools.yang.common.RpcResultBuilder; import org.opendaylight.yangtools.yang.common.Uint8; import org.slf4j.Logger; import org.slf4j.LoggerFactory; public class SalFlowServiceImpl implements SalFlowService { private static final Logger LOG = LoggerFactory.getLogger(SalFlowServiceImpl.class); private static final Uint8 OFPTT_ALL = Uint8.MAX_VALUE; private final MultiLayerFlowService flowUpdate; private final MultiLayerFlowService flowAdd; private final MultiLayerFlowService flowRemove; private final SingleLayerFlowService flowAddMessage; private final SingleLayerFlowService flowUpdateMessage; private final SingleLayerFlowService flowRemoveMessage; private final DeviceContext deviceContext; public SalFlowServiceImpl(final RequestContextStack requestContextStack, final DeviceContext deviceContext, final ConvertorExecutor convertorExecutor) { this.deviceContext = deviceContext; flowRemove = new MultiLayerFlowService<>(requestContextStack, deviceContext, RemoveFlowOutput.class, convertorExecutor); flowAdd = new MultiLayerFlowService<>(requestContextStack, deviceContext, AddFlowOutput.class, convertorExecutor); flowUpdate = new MultiLayerFlowService<>(requestContextStack, deviceContext, UpdateFlowOutput.class, convertorExecutor); flowAddMessage = new SingleLayerFlowService<>(requestContextStack, deviceContext, AddFlowOutput.class); flowUpdateMessage = new SingleLayerFlowService<>(requestContextStack, deviceContext, UpdateFlowOutput.class); flowRemoveMessage = new SingleLayerFlowService<>(requestContextStack, deviceContext, RemoveFlowOutput.class); } @Override public ListenableFuture> addFlow(final AddFlowInput input) { final FlowRegistryKey flowRegistryKey = FlowRegistryKeyFactory.create(deviceContext.getDeviceInfo().getVersion(), input); final ListenableFuture> future; if (flowAddMessage.canUseSingleLayerSerialization()) { future = flowAddMessage.handleServiceCall(input); Futures.addCallback(future, new AddFlowCallback(input, flowRegistryKey), MoreExecutors.directExecutor()); } else { future = flowAdd.processFlowModInputBuilders(flowAdd.toFlowModInputs(input)); Futures.addCallback(future, new AddFlowCallback(input, flowRegistryKey), MoreExecutors.directExecutor()); } return future; } @Override public ListenableFuture> removeFlow(final RemoveFlowInput input) { final ListenableFuture> future; if (flowRemoveMessage.canUseSingleLayerSerialization()) { future = flowRemoveMessage.handleServiceCall(input); Futures.addCallback(future, new RemoveFlowCallback(input), MoreExecutors.directExecutor()); } else { future = flowRemove.processFlowModInputBuilders(flowRemove.toFlowModInputs(input)); Futures.addCallback(future, new RemoveFlowCallback(input), MoreExecutors.directExecutor()); } return future; } @Override public ListenableFuture> updateFlow(final UpdateFlowInput input) { final UpdatedFlow updated = input.getUpdatedFlow(); final OriginalFlow original = input.getOriginalFlow(); final List allFlowMods = new ArrayList<>(); final List ofFlowModInputs; ListenableFuture> future; if (flowUpdateMessage.canUseSingleLayerSerialization()) { if (!FlowCreatorUtil.canModifyFlow(original, updated, flowUpdateMessage.getVersion())) { final SettableFuture> objectSettableFuture = SettableFuture.create(); final ListenableFuture>> listListenableFuture = Futures.successfulAsList(flowUpdateMessage.handleServiceCall(input.getOriginalFlow()), flowUpdateMessage.handleServiceCall(input.getUpdatedFlow())); Futures.addCallback(listListenableFuture, new FutureCallback>>() { @Override public void onSuccess(final List> results) { final ArrayList errors = new ArrayList<>(); for (RpcResult flowModResult : results) { if (flowModResult == null) { errors.add(RpcResultBuilder.newError( ErrorType.PROTOCOL, OFConstants.APPLICATION_TAG, "unexpected flowMod result (null) occurred")); } else if (!flowModResult.isSuccessful()) { errors.addAll(flowModResult.getErrors()); } } final RpcResultBuilder rpcResultBuilder; if (errors.isEmpty()) { rpcResultBuilder = RpcResultBuilder.success(); } else { rpcResultBuilder = RpcResultBuilder.failed().withRpcErrors(errors); } objectSettableFuture.set(rpcResultBuilder.build()); } @Override public void onFailure(final Throwable throwable) { RpcResultBuilder rpcResultBuilder = RpcResultBuilder.failed(); objectSettableFuture.set(rpcResultBuilder.build()); } }, MoreExecutors.directExecutor()); future = objectSettableFuture; } else { future = flowUpdateMessage.handleServiceCall(input.getUpdatedFlow()); } } else { if (!FlowCreatorUtil.canModifyFlow(original, updated, flowUpdate.getVersion())) { // We would need to remove original and add updated. // remove flow final RemoveFlowInputBuilder removeflow = new RemoveFlowInputBuilder(original); final List ofFlowRemoveInput = flowUpdate.toFlowModInputs(removeflow.build()); // remove flow should be the first allFlowMods.addAll(ofFlowRemoveInput); final AddFlowInputBuilder addFlowInputBuilder = new AddFlowInputBuilder(updated); ofFlowModInputs = flowUpdate.toFlowModInputs(addFlowInputBuilder.build()); } else { ofFlowModInputs = flowUpdate.toFlowModInputs(updated); } allFlowMods.addAll(ofFlowModInputs); future = flowUpdate.processFlowModInputBuilders(allFlowMods); } Futures.addCallback(future, new UpdateFlowCallback(input), MoreExecutors.directExecutor()); return future; } private final class AddFlowCallback implements FutureCallback> { private final AddFlowInput input; private final FlowRegistryKey flowRegistryKey; private AddFlowCallback(final AddFlowInput input, final FlowRegistryKey flowRegistryKey) { this.input = input; this.flowRegistryKey = flowRegistryKey; } @Override public void onSuccess(final RpcResult rpcResult) { if (!rpcResult.isSuccessful()) { if (LOG.isDebugEnabled()) { LOG.debug("Flow add failed for flow={}, errors={}", input, ErrorUtil.errorsToString(rpcResult.getErrors())); } return; } final DeviceFlowRegistry flowRegistry = deviceContext.getDeviceFlowRegistry(); final FlowDescriptor flowDescriptor; final FlowRef flowRef = input.getFlowRef(); if (flowRef != null) { final Uint8 tableId = input.getTableId(); final FlowId flowId = flowRef.getValue().firstKeyOf(Flow.class).getId(); flowDescriptor = FlowDescriptorFactory.create(tableId, flowId); // FIXME: this looks like an atomic operation flowRegistry.appendHistoryFlow(flowId, tableId, FlowGroupStatus.ADDED); flowRegistry.storeDescriptor(flowRegistryKey, flowDescriptor); } else { // FIXME: this looks like an atomic operation flowRegistry.store(flowRegistryKey); flowDescriptor = flowRegistry.retrieveDescriptor(flowRegistryKey); } if (LOG.isDebugEnabled()) { LOG.debug("Flow add with id={} finished without error", flowDescriptor.getFlowId().getValue()); } } @Override public void onFailure(final Throwable throwable) { LOG.warn("Service call for adding flow={} failed", input, throwable); } } private final class RemoveFlowCallback implements FutureCallback> { private final RemoveFlowInput input; private RemoveFlowCallback(final RemoveFlowInput input) { this.input = input; } @Override public void onSuccess(final RpcResult result) { if (result.isSuccessful()) { if (LOG.isDebugEnabled()) { LOG.debug("Flow remove finished without error for flow={}", input); } final DeviceFlowRegistry flowRegistry = deviceContext.getDeviceFlowRegistry(); if (input.getTableId() != null && !input.getTableId().equals(OFPTT_ALL)) { FlowRegistryKey flowRegistryKey = FlowRegistryKeyFactory.create(deviceContext.getDeviceInfo().getVersion(), input); flowRegistry.addMark(flowRegistryKey); final FlowRef flowRef = input.getFlowRef(); if (flowRef != null) { final FlowId flowId = flowRef.getValue().firstKeyOf(Flow.class).getId(); flowRegistry.appendHistoryFlow(flowId, input.getTableId(), FlowGroupStatus.REMOVED); } } else { flowRegistry.clearFlowRegistry(); } } else { if (LOG.isDebugEnabled()) { LOG.debug("Flow remove failed for flow={}, errors={}", input, ErrorUtil.errorsToString(result.getErrors())); } } } @Override public void onFailure(final Throwable throwable) { LOG.warn("Service call for removing flow={} failed", input, throwable); } } private final class UpdateFlowCallback implements FutureCallback> { private final UpdateFlowInput input; private UpdateFlowCallback(final UpdateFlowInput input) { this.input = input; } @Override public void onSuccess(final RpcResult updateFlowOutputRpcResult) { final DeviceFlowRegistry deviceFlowRegistry = deviceContext.getDeviceFlowRegistry(); final UpdatedFlow updated = input.getUpdatedFlow(); final OriginalFlow original = input.getOriginalFlow(); final FlowRegistryKey origFlowRegistryKey = FlowRegistryKeyFactory.create(deviceContext.getDeviceInfo().getVersion(), original); final FlowRegistryKey updatedFlowRegistryKey = FlowRegistryKeyFactory.create(deviceContext.getDeviceInfo().getVersion(), updated); final FlowDescriptor origFlowDescriptor = deviceFlowRegistry.retrieveDescriptor(origFlowRegistryKey); final boolean isUpdate = origFlowDescriptor != null; final FlowDescriptor updatedFlowDescriptor; final FlowRef flowRef = input.getFlowRef(); if (flowRef != null) { final Uint8 tableId = updated.getTableId(); final FlowId flowId = flowRef.getValue().firstKeyOf(Flow.class).getId(); // FIXME: this does not look right, we probably want better integration deviceFlowRegistry.appendHistoryFlow(flowId, tableId, FlowGroupStatus.MODIFIED); updatedFlowDescriptor = FlowDescriptorFactory.create(tableId, flowId); } else if (isUpdate) { updatedFlowDescriptor = origFlowDescriptor; } else { deviceFlowRegistry.store(updatedFlowRegistryKey); updatedFlowDescriptor = deviceFlowRegistry.retrieveDescriptor(updatedFlowRegistryKey); } if (isUpdate) { deviceFlowRegistry.addMark(origFlowRegistryKey); deviceFlowRegistry.storeDescriptor(updatedFlowRegistryKey, updatedFlowDescriptor); } } @Override public void onFailure(final Throwable throwable) { LOG.warn("Service call for updating flow={} failed", input, throwable); } } }