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=fba5816a548cc19419837f704744bbbe8fce61f5;hpb=b35e6899a297f86278772e48f6382e79bdde693d;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 fba5816a54..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.JdkFutureAdapters; import com.google.common.util.concurrent.ListenableFuture; -import com.google.common.util.concurrent.SettableFuture; -import java.math.BigInteger; import java.util.ArrayList; import java.util.Collection; -import java.util.Iterator; import java.util.List; -import java.util.concurrent.ExecutionException; import java.util.concurrent.Future; +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.device.Xid; +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.FlowHash; +import org.opendaylight.openflowplugin.api.openflow.registry.flow.FlowRegistryKey; +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.openflow.md.core.sal.convertor.FlowConvertor; +import org.opendaylight.openflowplugin.impl.registry.flow.FlowRegistryKeyFactory; +import org.opendaylight.openflowplugin.impl.util.FlowUtil; 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,64 +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.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.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); } - ListenableFuture> handleServiceCall(final BigInteger connectionID, - final FlowModInputBuilder flowModInputBuilder, final Function, ListenableFuture>> function) { - LOG.debug("Calling the FlowMod RPC method on MessageDispatchService"); - - final RequestContext requestContext = requestContextStack.createRequestContext(); - final SettableFuture> result = requestContextStack.storeOrFail(requestContext); - final DataCrate dataCrate = DataCrateBuilder.builder().setiDConnection(connectionID) - .setRequestContext(requestContext).setFlowModInputBuilder(flowModInputBuilder).build(); - - if (!result.isDone()) { - final ListenableFuture> resultFromOFLib = function.apply(dataCrate); - - final OFJResult2RequestCtxFuture OFJResult2RequestCtxFuture = new OFJResult2RequestCtxFuture<>(requestContext, deviceContext); - OFJResult2RequestCtxFuture.processResultFromOfJava(resultFromOFLib); - - } else { - RequestContextUtil.closeRequstContext(requestContext); - } - return result; + @Override + public void setItemLifecycleListener(@Nullable ItemLifecycleListener itemLifecycleListener) { + this.itemLifecycleListener = itemLifecycleListener; } @Override public Future> addFlow(final AddFlowInput input) { - - final FlowId flowId = input.getFlowRef().getValue().firstKeyOf(Flow.class, FlowKey.class).getId(); - - final FlowHash flowHash = FlowHashFactory.create(input); + final FlowId flowId; + if (null != input.getFlowRef()) { + flowId = input.getFlowRef().getValue().firstKeyOf(Flow.class, FlowKey.class).getId(); + } else { + flowId = FlowUtil.createAlienFlowId(input.getTableId()); + } + 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(flowHash, flowDescriptor); - - final List ofFlowModInputs = FlowConvertor.toFlowModInputs(input, version, datapathId); - final ListenableFuture future = processFlowModInputBuilders(ofFlowModInputs); - - Futures.addCallback(future, new FutureCallback() { + deviceContext.getDeviceFlowRegistry().store(flowRegistryKey, flowDescriptor); + final ListenableFuture> future = + flowAdd.processFlowModInputBuilders(flowAdd.toFlowModInputs(input)); + Futures.addCallback(future, new FutureCallback>() { @Override - public void onSuccess(final Object o) { - LOG.debug("flow add finished without error, id={}", flowId.getValue()); + public void onSuccess(final RpcResult rpcResult) { + if (rpcResult.isSuccessful()) { + 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.error("flow add failed for id={}, errors={}", flowId.getValue(), errorsToString(rpcResult.getErrors())); + } } @Override public void onFailure(final Throwable throwable) { - deviceContext.getDeviceFlowRegistry().markToBeremoved(flowHash); - LOG.trace("Service call for adding flows failed, id={}.", flowId.getValue(), throwable); + deviceContext.getDeviceFlowRegistry().markToBeremoved(flowRegistryKey); + LOG.error("Service call for adding flow with id={} failed, reason {} .", flowId.getValue(), throwable); } }); @@ -111,40 +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 this.handleServiceCall(PRIMARY_CONNECTION, - new Function, ListenableFuture>>() { - @Override - public ListenableFuture> apply(final DataCrate data) { - final FlowModInputBuilder ofFlowModInput = FlowConvertor.toFlowModInput(input, version, - datapathId); - final ListenableFuture> future = createResultForFlowMod(data, ofFlowModInput); - Futures.addCallback(future, new FutureCallback() { - @Override - public void onSuccess(final Object o) { - FlowHash flowHash = FlowHashFactory.create(input); - deviceContext.getDeviceFlowRegistry().markToBeremoved(flowHash); - } - @Override - public void onFailure(final Throwable throwable) { - StringBuffer errors = new StringBuffer(); - 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."); - } - LOG.trace("Flow modification failed. Errors : {}", errors.toString()); - } - }); - return future; + final ListenableFuture> future = + flowRemove.processFlowModInputBuilders(flowRemove.toFlowModInputs(input)); + Futures.addCallback(future, new FutureCallback>() { + @Override + public void onSuccess(final RpcResult result) { + if (result.isSuccessful()) { + if(LOG.isDebugEnabled()) { + LOG.debug("flow removed finished without error,"); + } + 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); + } } - }); + } 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 @@ -154,115 +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, version)) { + 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(), - version, datapathId); + 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(), version, datapathId); + ofFlowModInputs = flowUpdate.toFlowModInputs(addFlowInputBuilder.build()); } else { - ofFlowModInputs = FlowConvertor.toFlowModInputs(updated, version, datapathId); + ofFlowModInputs = flowUpdate.toFlowModInputs(updated); } allFlowMods.addAll(ofFlowModInputs); - ListenableFuture future = processFlowModInputBuilders(allFlowMods); - Futures.addCallback(future, new FutureCallback() { - @Override - public void onSuccess(final Object o) { - FlowHash flowHash = FlowHashFactory.create(original); - deviceContext.getDeviceFlowRegistry().markToBeremoved(flowHash); - - flowHash = FlowHashFactory.create(updated); - FlowId flowId = input.getFlowRef().getValue().firstKeyOf(Flow.class, FlowKey.class).getId(); - FlowDescriptor flowDescriptor = FlowDescriptorFactory.create(updated.getTableId(), flowId); - deviceContext.getDeviceFlowRegistry().store(flowHash, flowDescriptor); - - } - + ListenableFuture> future = flowUpdate.processFlowModInputBuilders(allFlowMods); + Futures.addCallback(future, new FutureCallback>() { @Override - public void onFailure(final Throwable throwable) { - - } - }); - return future; - } - - private ListenableFuture> processFlowModInputBuilders( - final List ofFlowModInputs) { - final List>> partialFutures = new ArrayList<>(); - for (FlowModInputBuilder flowModInputBuilder : ofFlowModInputs) { - ListenableFuture> partialFuture = handleServiceCall(PRIMARY_CONNECTION, flowModInputBuilder, - new Function, ListenableFuture>>() { - @Override - public ListenableFuture> apply(final DataCrate data) { - return createResultForFlowMod(data); - } - }); - partialFutures.add(partialFuture); - } - - final ListenableFuture>> allFutures = Futures.allAsList(partialFutures); - final SettableFuture> finalFuture = SettableFuture.create(); - Futures.addCallback(allFutures, new FutureCallback>>() { - @Override - public void onSuccess(List> results) { - Iterator flowModInputBldIterator = ofFlowModInputs.iterator(); - List rpcErrorLot = new ArrayList<>(); - for (RpcResult rpcResult : results) { - FlowModInputBuilder flowModInputBld = flowModInputBldIterator.next(); - if (rpcResult.isSuccessful()) { - Long xid = flowModInputBld.getXid(); - LOG.warn("Positive confirmation of flow push is not supported by OF-spec"); - LOG.warn("flow future result was successful [{}] = this should have never happen", - xid); - rpcErrorLot.add(RpcResultBuilder.newError(ErrorType.APPLICATION, "", - "flow future result was successful ["+xid+"] = this should have never happen")); - } else { - rpcErrorLot.addAll(rpcResult.getErrors()); + public void onSuccess(final RpcResult o) { + FlowRegistryKey flowRegistryKey = FlowRegistryKeyFactory.create(original); + + FlowRegistryKey updatedflowRegistryKey = FlowRegistryKeyFactory.create(updated); + final FlowRef flowRef = input.getFlowRef(); + final DeviceFlowRegistry deviceFlowRegistry = deviceContext.getDeviceFlowRegistry(); + deviceFlowRegistry.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); + } + } + //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()); } } - finalFuture.set(RpcResultBuilder.failed().withRpcErrors(rpcErrorLot).build()); } @Override - public void onFailure(Throwable t) { - LOG.trace("Flow mods chained future failed."); - RpcResultBuilder resultBuilder; - if (allFutures.isCancelled()) { - if (LOG.isTraceEnabled()) { - for (FlowModInputBuilder ofFlowModInput : ofFlowModInputs) { - LOG.trace("flow future result was cancelled [{}] = barrier passed it without error", - ofFlowModInput.getXid()); - } - } - resultBuilder = RpcResultBuilder.success(); - } else { - resultBuilder = RpcResultBuilder.failed().withError(ErrorType.APPLICATION, "", t.getMessage()); - } - finalFuture.set(resultBuilder.build()); + public void onFailure(final Throwable throwable) { + LOG.error("Service call for updating flow failed, reason{}", throwable); } }); - - return finalFuture; - } - - protected ListenableFuture> createResultForFlowMod(final DataCrate data) { - return createResultForFlowMod(data, data.getFlowModInputBuilder()); + return future; } - protected ListenableFuture> createResultForFlowMod(final DataCrate data, final FlowModInputBuilder flowModInput) { - final Xid xid = data.getRequestContext().getXid(); - flowModInput.setXid(xid.getValue()); - Future> flowModResult = provideConnectionAdapter(data.getiDConnection()).flowMod( - flowModInput.build()); - return JdkFutureAdapters.listenInPoolThread(flowModResult); + @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())); } - }