X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=openflowplugin-impl%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fopenflowplugin%2Fimpl%2Fservices%2FSalFlowServiceImpl.java;h=cf2a160998321f1077b310cbbf79b142acebecb7;hb=6e2627630ebabb37526133a13deeb62adcc85d35;hp=aa6b59e884070b9c574a7a257113122cf4e79fc5;hpb=0a8ff1abfd66739bc3c8ad3322cf5466eff40627;p=openflowplugin.git diff --git a/openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/services/SalFlowServiceImpl.java b/openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/services/SalFlowServiceImpl.java index aa6b59e884..cf2a160998 100644 --- a/openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/services/SalFlowServiceImpl.java +++ b/openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/services/SalFlowServiceImpl.java @@ -1,37 +1,37 @@ /** * 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; -import com.google.common.base.Function; +import com.google.common.annotations.VisibleForTesting; 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.SettableFuture; import java.util.ArrayList; import java.util.Collection; import java.util.List; -import java.util.concurrent.ExecutionException; import java.util.concurrent.Future; -import org.opendaylight.openflowjava.protocol.api.connection.OutboundQueue; +import javax.annotation.Nullable; import org.opendaylight.openflowplugin.api.openflow.device.DeviceContext; -import org.opendaylight.openflowplugin.api.openflow.device.RequestContext; 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.api.openflow.statistics.ofpspecific.MessageSpy; +import org.opendaylight.openflowplugin.api.openflow.rpc.ItemLifeCycleSource; +import org.opendaylight.openflowplugin.api.openflow.rpc.listener.ItemLifecycleListener; import org.opendaylight.openflowplugin.impl.registry.flow.FlowDescriptorFactory; -import org.opendaylight.openflowplugin.impl.registry.flow.FlowHashFactory; +import org.opendaylight.openflowplugin.impl.registry.flow.FlowRegistryKeyFactory; import org.opendaylight.openflowplugin.impl.util.FlowUtil; -import org.opendaylight.openflowplugin.openflow.md.core.sal.convertor.FlowConvertor; import org.opendaylight.openflowplugin.openflow.md.util.FlowCreatorUtil; +import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode; 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; import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.Flow; +import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.FlowBuilder; import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.FlowKey; import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.AddFlowInput; import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.AddFlowInputBuilder; @@ -44,56 +44,72 @@ import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.Upda 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.openflow.protocol.rev130731.FlowModInput; +import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.types.rev131026.FlowRef; +import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node; +import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.NodeKey; import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.FlowModInputBuilder; -import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.OfHeader; +import org.opendaylight.yangtools.yang.binding.KeyedInstanceIdentifier; import org.opendaylight.yangtools.yang.common.RpcError; -import org.opendaylight.yangtools.yang.common.RpcError.ErrorType; import org.opendaylight.yangtools.yang.common.RpcResult; -import org.opendaylight.yangtools.yang.common.RpcResultBuilder; import org.slf4j.Logger; +import org.slf4j.LoggerFactory; -public class SalFlowServiceImpl extends CommonService implements SalFlowService { - - private static final Logger LOG = org.slf4j.LoggerFactory.getLogger(SalFlowServiceImpl.class); +public class SalFlowServiceImpl implements SalFlowService, ItemLifeCycleSource { + private static final Logger LOG = LoggerFactory.getLogger(SalFlowServiceImpl.class); + private final FlowService flowUpdate; + private final FlowService flowAdd; + private final FlowService flowRemove; + private final DeviceContext deviceContext; + private ItemLifecycleListener itemLifecycleListener; public SalFlowServiceImpl(final RequestContextStack requestContextStack, final DeviceContext deviceContext) { - super(requestContextStack, deviceContext); + this.deviceContext = deviceContext; + flowRemove = new FlowService(requestContextStack, deviceContext, RemoveFlowOutput.class); + flowAdd = new FlowService<>(requestContextStack, deviceContext, AddFlowOutput.class); + flowUpdate = new FlowService<>(requestContextStack, deviceContext, UpdateFlowOutput.class); } @Override - public Future> addFlow(final AddFlowInput input) { - getMessageSpy().spyMessage(input.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.TO_SWITCH_ENTERED); + public void setItemLifecycleListener(@Nullable ItemLifecycleListener itemLifecycleListener) { + this.itemLifecycleListener = itemLifecycleListener; + } - final List ofFlowModInputs = FlowConvertor.toFlowModInputs(input, getVersion(), getDatapathId()); - final ListenableFuture> future = processFlowModInputBuilders(ofFlowModInputs); + @Override + public Future> addFlow(final AddFlowInput input) { final FlowId flowId; if (null != input.getFlowRef()) { flowId = input.getFlowRef().getValue().firstKeyOf(Flow.class, FlowKey.class).getId(); } else { flowId = FlowUtil.createAlienFlowId(input.getTableId()); } - - final DeviceContext deviceContext = getDeviceContext(); - final FlowRegistryKey flowRegistryKey = FlowHashFactory.create(input, deviceContext.getPrimaryConnectionContext().getFeatures().getVersion()); + LOG.trace("Calling add flow for flow with ID ={}.", flowId); + final FlowRegistryKey flowRegistryKey = FlowRegistryKeyFactory.create(input); final FlowDescriptor flowDescriptor = FlowDescriptorFactory.create(input.getTableId(), flowId); deviceContext.getDeviceFlowRegistry().store(flowRegistryKey, flowDescriptor); + final ListenableFuture> future = + flowAdd.processFlowModInputBuilders(flowAdd.toFlowModInputs(input)); Futures.addCallback(future, new FutureCallback>() { - - @Override public void onSuccess(final RpcResult rpcResult) { if (rpcResult.isSuccessful()) { - LOG.debug("flow add finished without error, id={}", flowId.getValue()); + if(LOG.isDebugEnabled()) { + LOG.debug("flow add with id={},finished without error,", flowId.getValue()); + } + if (itemLifecycleListener != null) { + KeyedInstanceIdentifier flowPath = createFlowPath(flowDescriptor, + deviceContext.getDeviceInfo().getNodeInstanceIdentifier()); + final FlowBuilder flowBuilder = new FlowBuilder(input).setId(flowDescriptor.getFlowId()); + itemLifecycleListener.onAdded(flowPath, flowBuilder.build()); + } } else { - LOG.debug("flow add failed with error, id={}", flowId.getValue()); - } + LOG.error("flow add failed for id={}, errors={}", flowId.getValue(), errorsToString(rpcResult.getErrors())); + } } @Override public void onFailure(final Throwable throwable) { deviceContext.getDeviceFlowRegistry().markToBeremoved(flowRegistryKey); - LOG.trace("Service call for adding flows failed, id={}.", flowId.getValue(), throwable); + LOG.error("Service call for adding flow with id={} failed, reason {} .", flowId.getValue(), throwable); } }); @@ -103,44 +119,49 @@ public class SalFlowServiceImpl extends CommonService implements SalFlowService @Override public Future> removeFlow(final RemoveFlowInput input) { LOG.trace("Calling remove flow for flow with ID ={}.", input.getFlowRef()); - return handleServiceCall(new Function, ListenableFuture>>() { + + final ListenableFuture> future = + flowRemove.processFlowModInputBuilders(flowRemove.toFlowModInputs(input)); + Futures.addCallback(future, new FutureCallback>() { @Override - public ListenableFuture> apply(final RequestContext requestContext) { - final FlowModInputBuilder ofFlowModInput = FlowConvertor.toFlowModInput(input, getVersion(), - getDatapathId()); - final ListenableFuture> future = createResultForFlowMod(requestContext, ofFlowModInput); - Futures.addCallback(future, new FutureCallback>() { - @Override - public void onSuccess(final RpcResult o) { - final DeviceContext deviceContext = getDeviceContext(); - getMessageSpy().spyMessage(input.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.TO_SWITCH_SUBMIT_SUCCESS); - FlowRegistryKey flowRegistryKey = FlowHashFactory.create(input, deviceContext.getPrimaryConnectionContext().getFeatures().getVersion()); - deviceContext.getDeviceFlowRegistry().markToBeremoved(flowRegistryKey); + public void onSuccess(final RpcResult result) { + if (result.isSuccessful()) { + if(LOG.isDebugEnabled()) { + LOG.debug("flow removed finished without error,"); } - - @Override - public void onFailure(final Throwable throwable) { - getMessageSpy().spyMessage(input.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.TO_SWITCH_SUBMIT_FAILURE); - LOG.trace("Flow modification failed..", throwable); - StringBuilder errors = new StringBuilder(); - try { - RpcResult result = future.get(); - Collection rpcErrors = result.getErrors(); - if (null != rpcErrors && rpcErrors.size() > 0) { - for (RpcError rpcError : rpcErrors) { - errors.append(rpcError.getMessage()); - } - } - } catch (InterruptedException | ExecutionException e) { - LOG.trace("Flow modification failed. Can't read errors from RpcResult."); - } finally { - LOG.trace("Flow modification failed. Errors : {}", errors.toString()); + FlowRegistryKey flowRegistryKey = FlowRegistryKeyFactory.create(input); + deviceContext.getDeviceFlowRegistry().markToBeremoved(flowRegistryKey); + if (itemLifecycleListener != null) { + final FlowDescriptor flowDescriptor = + deviceContext.getDeviceFlowRegistry().retrieveIdForFlow(flowRegistryKey); + if (flowDescriptor != null) { + KeyedInstanceIdentifier flowPath = createFlowPath(flowDescriptor, + deviceContext.getDeviceInfo().getNodeInstanceIdentifier()); + itemLifecycleListener.onRemoved(flowPath); } } - }); - return future; + } else { + LOG.error("Flow remove failed with errors : {}",errorsToString(result.getErrors())); + } + } + + @Override + public void onFailure(final Throwable throwable) { + LOG.error("Service call for removing flow with id {} failed ,reason {}",input.getFlowRef().getValue(), throwable); } }); + + return future; + } + + private final String errorsToString(final Collection rpcErrors) { + final StringBuilder errors = new StringBuilder(); + if ((null != rpcErrors) && (rpcErrors.size() > 0)) { + for (final RpcError rpcError : rpcErrors) { + errors.append(rpcError.getMessage()); + } + } + return errors.toString(); } @Override @@ -150,110 +171,70 @@ public class SalFlowServiceImpl extends CommonService implements SalFlowService final OriginalFlow original = in.getOriginalFlow(); final List allFlowMods = new ArrayList<>(); - List ofFlowModInputs; + final List ofFlowModInputs; - if (!FlowCreatorUtil.canModifyFlow(original, updated, getVersion())) { + 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 = FlowConvertor.toFlowModInputs(removeflow.build(), - getVersion(), getDatapathId()); + final List ofFlowRemoveInput = flowUpdate.toFlowModInputs(removeflow.build()); // remove flow should be the first allFlowMods.addAll(ofFlowRemoveInput); final AddFlowInputBuilder addFlowInputBuilder = new AddFlowInputBuilder(updated); - ofFlowModInputs = FlowConvertor.toFlowModInputs(addFlowInputBuilder.build(), getVersion(), getDatapathId()); + ofFlowModInputs = flowUpdate.toFlowModInputs(addFlowInputBuilder.build()); } else { - ofFlowModInputs = FlowConvertor.toFlowModInputs(updated, getVersion(), getDatapathId()); + ofFlowModInputs = flowUpdate.toFlowModInputs(updated); } allFlowMods.addAll(ofFlowModInputs); - ListenableFuture> future = processFlowModInputBuilders(allFlowMods); + ListenableFuture> future = flowUpdate.processFlowModInputBuilders(allFlowMods); Futures.addCallback(future, new FutureCallback>() { @Override public void onSuccess(final RpcResult o) { - final DeviceContext deviceContext = getDeviceContext(); - getMessageSpy().spyMessage(input.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.TO_SWITCH_SUBMIT_SUCCESS); - final short version = deviceContext.getPrimaryConnectionContext().getFeatures().getVersion(); - FlowRegistryKey flowRegistryKey = FlowHashFactory.create(original, version); + FlowRegistryKey flowRegistryKey = FlowRegistryKeyFactory.create(original); - FlowRegistryKey updatedflowRegistryKey = FlowHashFactory.create(updated, version); - FlowId flowId = input.getFlowRef().getValue().firstKeyOf(Flow.class, FlowKey.class).getId(); - FlowDescriptor flowDescriptor = FlowDescriptorFactory.create(updated.getTableId(), flowId); + FlowRegistryKey updatedflowRegistryKey = FlowRegistryKeyFactory.create(updated); + final FlowRef flowRef = input.getFlowRef(); final DeviceFlowRegistry deviceFlowRegistry = deviceContext.getDeviceFlowRegistry(); deviceFlowRegistry.markToBeremoved(flowRegistryKey); - deviceFlowRegistry.store(updatedflowRegistryKey, flowDescriptor); + + if (itemLifecycleListener != null) { + final FlowDescriptor flowDescriptor = deviceContext.getDeviceFlowRegistry().retrieveIdForFlow(flowRegistryKey); + if (flowDescriptor != null) { + KeyedInstanceIdentifier flowPath = createFlowPath(flowDescriptor, + deviceContext.getDeviceInfo().getNodeInstanceIdentifier()); + itemLifecycleListener.onRemoved(flowPath); + } + } + //if provided, store flow id to flow registry + if (flowRef != null) { + final FlowId flowId = flowRef.getValue().firstKeyOf(Flow.class, FlowKey.class).getId(); + final FlowDescriptor flowDescriptor = FlowDescriptorFactory.create(updated.getTableId(), flowId); + deviceFlowRegistry.store(updatedflowRegistryKey, flowDescriptor); + + if (itemLifecycleListener != null) { + KeyedInstanceIdentifier flowPath = createFlowPath(flowDescriptor, + deviceContext.getDeviceInfo().getNodeInstanceIdentifier()); + final FlowBuilder flowBuilder = new FlowBuilder(input.getUpdatedFlow()).setId(flowDescriptor.getFlowId()); + itemLifecycleListener.onAdded(flowPath, flowBuilder.build()); + } + } } @Override public void onFailure(final Throwable throwable) { - getMessageSpy().spyMessage(input.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.TO_SWITCH_SUBMIT_FAILURE); + LOG.error("Service call for updating flow failed, reason{}", throwable); } }); return future; } - private ListenableFuture> processFlowModInputBuilders(final List ofFlowModInputs) { - - final List>> partialFutures = new ArrayList<>(); - - for (final FlowModInputBuilder flowModInputBuilder : ofFlowModInputs) { - ListenableFuture> partialFuture = handleServiceCall( - new Function, ListenableFuture>>() { - @Override - public ListenableFuture> apply(final RequestContext requestContext) { - return createResultForFlowMod(requestContext, flowModInputBuilder); - } - }); - partialFutures.add(partialFuture); - } - - final ListenableFuture>> allFutures = Futures.successfulAsList(partialFutures); - final SettableFuture> finalFuture = SettableFuture.create(); - Futures.addCallback(allFutures, new FutureCallback>>() { - @Override - public void onSuccess(final List> results) { - RpcResultBuilder rpcResultBuilder = RpcResultBuilder.success(); - finalFuture.set(rpcResultBuilder.build()); - } - - @Override - public void onFailure(final Throwable t) { - RpcResultBuilder rpcResultBuilder = RpcResultBuilder.failed(); - finalFuture.set(rpcResultBuilder.build()); - } - }); - - return finalFuture; - } - - protected ListenableFuture> createResultForFlowMod(final RequestContext requestContext, final FlowModInputBuilder flowModInputBuilder) { - final OutboundQueue outboundQueue = getDeviceContext().getPrimaryConnectionContext().getOutboundQueueProvider(); - final long xid = requestContext.getXid().getValue(); - flowModInputBuilder.setXid(xid); - final FlowModInput flowModInput = flowModInputBuilder.build(); - - outboundQueue.commitEntry(xid, flowModInput, new FutureCallback() { - @Override - public void onSuccess(final OfHeader ofHeader) { - getMessageSpy().spyMessage(FlowModInput.class, MessageSpy.STATISTIC_GROUP.TO_SWITCH_SUBMIT_SUCCESS); - - RpcResultBuilder rpcResultBuilder = RpcResultBuilder.success(); - requestContext.setResult(rpcResultBuilder.build()); - - RequestContextUtil.closeRequstContext(requestContext); - } - - @Override - public void onFailure(final Throwable throwable) { - getMessageSpy().spyMessage(FlowModInput.class, MessageSpy.STATISTIC_GROUP.TO_SWITCH_SUBMIT_FAILURE); - RpcResultBuilder rpcResultBuilder = RpcResultBuilder.failed().withError(ErrorType.APPLICATION, throwable.getMessage(), throwable); - requestContext.setResult(rpcResultBuilder.build()); - - RequestContextUtil.closeRequstContext(requestContext); - } - }); - return requestContext.getFuture(); + @VisibleForTesting + static KeyedInstanceIdentifier createFlowPath(FlowDescriptor flowDescriptor, + KeyedInstanceIdentifier nodePath) { + return nodePath.augmentation(FlowCapableNode.class) + .child(Table.class, flowDescriptor.getTableKey()) + .child(Flow.class, new FlowKey(flowDescriptor.getFlowId())); } - }