2 * Copyright (c) 2015, 2017 Cisco Systems, Inc. and others. All rights reserved.
4 * This program and the accompanying materials are made available under the
5 * terms of the Eclipse Public License v1.0 which accompanies this distribution,
6 * and is available at http://www.eclipse.org/legal/epl-v10.html
8 package org.opendaylight.openflowplugin.applications.bulk.o.matic;
10 import static java.util.Objects.requireNonNull;
11 import static java.util.Objects.requireNonNullElse;
13 import com.google.common.util.concurrent.FluentFuture;
14 import com.google.common.util.concurrent.FutureCallback;
15 import com.google.common.util.concurrent.Futures;
16 import com.google.common.util.concurrent.ListenableFuture;
17 import com.google.common.util.concurrent.MoreExecutors;
18 import com.google.common.util.concurrent.SettableFuture;
19 import java.lang.management.ManagementFactory;
20 import java.util.ArrayList;
21 import java.util.Collections;
22 import java.util.List;
23 import java.util.concurrent.ExecutorService;
24 import java.util.concurrent.ForkJoinPool;
25 import javax.management.InstanceAlreadyExistsException;
26 import javax.management.MBeanRegistrationException;
27 import javax.management.MBeanServer;
28 import javax.management.MalformedObjectNameException;
29 import javax.management.NotCompliantMBeanException;
30 import javax.management.ObjectName;
31 import org.opendaylight.infrautils.utils.concurrent.LoggingFutures;
32 import org.opendaylight.mdsal.binding.api.DataBroker;
33 import org.opendaylight.mdsal.binding.api.WriteTransaction;
34 import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
35 import org.opendaylight.yang.gen.v1.urn.opendaylight.bulk.flow.service.rev150608.AddFlowsDsInput;
36 import org.opendaylight.yang.gen.v1.urn.opendaylight.bulk.flow.service.rev150608.AddFlowsDsOutput;
37 import org.opendaylight.yang.gen.v1.urn.opendaylight.bulk.flow.service.rev150608.AddFlowsRpcInput;
38 import org.opendaylight.yang.gen.v1.urn.opendaylight.bulk.flow.service.rev150608.AddFlowsRpcOutput;
39 import org.opendaylight.yang.gen.v1.urn.opendaylight.bulk.flow.service.rev150608.BulkFlowBaseContentGrouping;
40 import org.opendaylight.yang.gen.v1.urn.opendaylight.bulk.flow.service.rev150608.FlowRpcAddMultipleInput;
41 import org.opendaylight.yang.gen.v1.urn.opendaylight.bulk.flow.service.rev150608.FlowRpcAddMultipleOutput;
42 import org.opendaylight.yang.gen.v1.urn.opendaylight.bulk.flow.service.rev150608.FlowRpcAddTestInput;
43 import org.opendaylight.yang.gen.v1.urn.opendaylight.bulk.flow.service.rev150608.FlowRpcAddTestOutput;
44 import org.opendaylight.yang.gen.v1.urn.opendaylight.bulk.flow.service.rev150608.FlowTestInput;
45 import org.opendaylight.yang.gen.v1.urn.opendaylight.bulk.flow.service.rev150608.FlowTestOutput;
46 import org.opendaylight.yang.gen.v1.urn.opendaylight.bulk.flow.service.rev150608.ReadFlowTestInput;
47 import org.opendaylight.yang.gen.v1.urn.opendaylight.bulk.flow.service.rev150608.ReadFlowTestOutput;
48 import org.opendaylight.yang.gen.v1.urn.opendaylight.bulk.flow.service.rev150608.RegisterInput;
49 import org.opendaylight.yang.gen.v1.urn.opendaylight.bulk.flow.service.rev150608.RegisterInputBuilder;
50 import org.opendaylight.yang.gen.v1.urn.opendaylight.bulk.flow.service.rev150608.RegisterOutput;
51 import org.opendaylight.yang.gen.v1.urn.opendaylight.bulk.flow.service.rev150608.RemoveFlowsDsInput;
52 import org.opendaylight.yang.gen.v1.urn.opendaylight.bulk.flow.service.rev150608.RemoveFlowsDsOutput;
53 import org.opendaylight.yang.gen.v1.urn.opendaylight.bulk.flow.service.rev150608.RemoveFlowsRpcInput;
54 import org.opendaylight.yang.gen.v1.urn.opendaylight.bulk.flow.service.rev150608.RemoveFlowsRpcOutput;
55 import org.opendaylight.yang.gen.v1.urn.opendaylight.bulk.flow.service.rev150608.SalBulkFlowService;
56 import org.opendaylight.yang.gen.v1.urn.opendaylight.bulk.flow.service.rev150608.TableTestInput;
57 import org.opendaylight.yang.gen.v1.urn.opendaylight.bulk.flow.service.rev150608.TableTestOutput;
58 import org.opendaylight.yang.gen.v1.urn.opendaylight.bulk.flow.service.rev150608.bulk.flow.ds.list.grouping.BulkFlowDsItem;
59 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode;
60 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowId;
61 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.Table;
62 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.TableKey;
63 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.Flow;
64 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.FlowBuilder;
65 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.FlowKey;
66 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.AddFlowInputBuilder;
67 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.AddFlowOutput;
68 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.RemoveFlowInputBuilder;
69 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.RemoveFlowOutput;
70 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.SalFlowService;
71 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeRef;
72 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
73 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
74 import org.opendaylight.yangtools.yang.common.RpcError;
75 import org.opendaylight.yangtools.yang.common.RpcResult;
76 import org.opendaylight.yangtools.yang.common.RpcResultBuilder;
77 import org.slf4j.Logger;
78 import org.slf4j.LoggerFactory;
81 * Simple implementation providing bulk flows operations.
83 public class SalBulkFlowServiceImpl implements SalBulkFlowService {
85 private static final Logger LOG = LoggerFactory.getLogger(SalBulkFlowServiceImpl.class);
87 private final SalFlowService flowService;
88 private final DataBroker dataBroker;
89 private final FlowCounter flowCounterBeanImpl = new FlowCounter();
90 private final ExecutorService fjService = new ForkJoinPool();
92 public SalBulkFlowServiceImpl(final SalFlowService flowService, final DataBroker dataBroker) {
93 this.flowService = requireNonNull(flowService);
94 this.dataBroker = requireNonNull(dataBroker);
96 LoggingFutures.addErrorLogging(register(new RegisterInputBuilder().build()), LOG, "register");
100 public ListenableFuture<RpcResult<AddFlowsDsOutput>> addFlowsDs(final AddFlowsDsInput input) {
101 WriteTransaction writeTransaction = dataBroker.newWriteOnlyTransaction();
102 boolean createParentsNextTime = requireNonNullElse(input.isAlwaysCreateParents(), Boolean.FALSE);
103 boolean createParents = true;
104 for (BulkFlowDsItem bulkFlow : input.getBulkFlowDsItem()) {
105 FlowBuilder flowBuilder = new FlowBuilder(bulkFlow);
106 flowBuilder.setTableId(bulkFlow.getTableId());
107 flowBuilder.setId(new FlowId(bulkFlow.getFlowId()));
109 writeTransaction.mergeParentStructurePut(LogicalDatastoreType.CONFIGURATION,
110 getFlowInstanceIdentifier(bulkFlow),
111 flowBuilder.build());
113 writeTransaction.put(LogicalDatastoreType.CONFIGURATION, getFlowInstanceIdentifier(bulkFlow),
114 flowBuilder.build());
116 createParents = createParentsNextTime;
118 FluentFuture<?> submitFuture = writeTransaction.commit();
119 return Futures.transform(handleResultFuture(Futures.allAsList(submitFuture)), voidRpcResult -> {
120 if (voidRpcResult.isSuccessful()) {
121 return RpcResultBuilder.<AddFlowsDsOutput>success().build();
123 return RpcResultBuilder.<AddFlowsDsOutput>failed().build();
125 },MoreExecutors.directExecutor());
128 private static InstanceIdentifier<Flow> getFlowInstanceIdentifier(final BulkFlowDsItem bulkFlow) {
129 final NodeRef nodeRef = bulkFlow.getNode();
130 return ((InstanceIdentifier<Node>) nodeRef.getValue()).augmentation(FlowCapableNode.class)
131 .child(Table.class, new TableKey(bulkFlow.getTableId()))
132 .child(Flow.class, new FlowKey(new FlowId(bulkFlow.getFlowId())));
136 public ListenableFuture<RpcResult<RemoveFlowsDsOutput>> removeFlowsDs(final RemoveFlowsDsInput input) {
137 WriteTransaction writeTransaction = dataBroker.newWriteOnlyTransaction();
138 for (BulkFlowDsItem bulkFlow : input.getBulkFlowDsItem()) {
139 writeTransaction.delete(LogicalDatastoreType.CONFIGURATION, getFlowInstanceIdentifier(bulkFlow));
141 return Futures.transform(handleResultFuture(Futures.allAsList(writeTransaction.commit())), voidRpcResult -> {
142 if (voidRpcResult.isSuccessful()) {
143 return RpcResultBuilder.<RemoveFlowsDsOutput>success().build();
145 return RpcResultBuilder.<RemoveFlowsDsOutput>failed().build();
147 }, MoreExecutors.directExecutor());
150 private static <T> ListenableFuture<RpcResult<Void>> handleResultFuture(
151 final ListenableFuture<List<T>> submitFuture) {
152 final SettableFuture<RpcResult<Void>> rpcResult = SettableFuture.create();
153 Futures.addCallback(submitFuture, new FutureCallback<List<T>>() {
155 public void onSuccess(final List<T> result) {
156 rpcResult.set(RpcResultBuilder.success((Void) null).build());
160 public void onFailure(final Throwable throwable) {
161 RpcResultBuilder<Void> rpcResultBld = RpcResultBuilder.<Void>failed()
162 .withRpcErrors(Collections.singleton(RpcResultBuilder.newError(RpcError.ErrorType.APPLICATION,
163 null, throwable.getMessage())));
164 rpcResult.set(rpcResultBld.build());
166 }, MoreExecutors.directExecutor());
171 public ListenableFuture<RpcResult<AddFlowsRpcOutput>> addFlowsRpc(final AddFlowsRpcInput input) {
172 List<ListenableFuture<RpcResult<AddFlowOutput>>> bulkResults = new ArrayList<>();
174 for (BulkFlowBaseContentGrouping bulkFlow : input.getBulkFlowItem()) {
175 AddFlowInputBuilder flowInputBuilder = new AddFlowInputBuilder(
176 (org.opendaylight.yang.gen.v1.urn.opendaylight.flow.types.rev131026.Flow) bulkFlow);
177 final NodeRef nodeRef = bulkFlow.getNode();
178 flowInputBuilder.setNode(nodeRef);
179 flowInputBuilder.setTableId(bulkFlow.getTableId());
180 bulkResults.add(flowService.addFlow(flowInputBuilder.build()));
182 return Futures.transform(handleResultFuture(Futures.allAsList(bulkResults)), voidRpcResult -> {
183 if (voidRpcResult.isSuccessful()) {
184 return RpcResultBuilder.<AddFlowsRpcOutput>success().build();
186 return RpcResultBuilder.<AddFlowsRpcOutput>failed().build();
188 },MoreExecutors.directExecutor());
192 public ListenableFuture<RpcResult<ReadFlowTestOutput>> readFlowTest(final ReadFlowTestInput input) {
193 FlowReader flowReader = FlowReader.getNewInstance(dataBroker, input.getDpnCount().intValue(),
194 input.getFlowsPerDpn().intValue(), input.isVerbose(), input.isIsConfigDs(),
195 input.getStartTableId().shortValue(), input.getEndTableId().shortValue());
196 flowCounterBeanImpl.setReader(flowReader);
197 fjService.execute(flowReader);
198 RpcResultBuilder<ReadFlowTestOutput> rpcResultBuilder = RpcResultBuilder.success();
199 return Futures.immediateFuture(rpcResultBuilder.build());
203 public ListenableFuture<RpcResult<FlowRpcAddTestOutput>> flowRpcAddTest(final FlowRpcAddTestInput input) {
204 FlowWriterDirectOFRpc flowAddRpcTestImpl = new FlowWriterDirectOFRpc(dataBroker, flowService, fjService);
205 flowAddRpcTestImpl.rpcFlowAdd(input.getDpnId(), input.getFlowCount().intValue(),
206 input.getRpcBatchSize().intValue());
208 RpcResultBuilder<FlowRpcAddTestOutput> rpcResultBuilder = RpcResultBuilder.success();
209 return Futures.immediateFuture(rpcResultBuilder.build());
213 public ListenableFuture<RpcResult<RegisterOutput>> register(final RegisterInput input) {
214 RpcResultBuilder<RegisterOutput> rpcResultBuilder = RpcResultBuilder.success();
216 MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
217 String pathToMBean = String.format("%s:type=%s", FlowCounter.class.getPackage().getName(),
218 FlowCounter.class.getSimpleName());
219 ObjectName name = new ObjectName(pathToMBean);
220 mbs.registerMBean(flowCounterBeanImpl, name);
221 } catch (MalformedObjectNameException | InstanceAlreadyExistsException | MBeanRegistrationException
222 | NotCompliantMBeanException e) {
223 rpcResultBuilder = RpcResultBuilder.failed();
224 LOG.warn("Exception occurred", e);
226 return Futures.immediateFuture(rpcResultBuilder.build());
230 public ListenableFuture<RpcResult<RemoveFlowsRpcOutput>> removeFlowsRpc(final RemoveFlowsRpcInput input) {
231 List<ListenableFuture<RpcResult<RemoveFlowOutput>>> bulkResults = new ArrayList<>();
233 for (BulkFlowBaseContentGrouping bulkFlow : input.getBulkFlowItem()) {
234 RemoveFlowInputBuilder flowInputBuilder = new RemoveFlowInputBuilder(
235 (org.opendaylight.yang.gen.v1.urn.opendaylight.flow.types.rev131026.Flow) bulkFlow);
236 final NodeRef nodeRef = bulkFlow.getNode();
237 flowInputBuilder.setNode(nodeRef);
238 flowInputBuilder.setTableId(bulkFlow.getTableId());
239 bulkResults.add(flowService.removeFlow(flowInputBuilder.build()));
241 return Futures.transform(handleResultFuture(Futures.allAsList(bulkResults)), voidRpcResult -> {
242 if (voidRpcResult.isSuccessful()) {
243 return RpcResultBuilder.<RemoveFlowsRpcOutput>success().build();
245 return RpcResultBuilder.<RemoveFlowsRpcOutput>failed().build();
247 }, MoreExecutors.directExecutor());
251 public ListenableFuture<RpcResult<FlowTestOutput>> flowTest(final FlowTestInput input) {
252 if (input.isTxChain()) {
253 FlowWriterTxChain flowTester = new FlowWriterTxChain(dataBroker, fjService);
254 flowCounterBeanImpl.setWriter(flowTester);
255 if (input.isIsAdd()) {
256 flowTester.addFlows(input.getDpnCount().intValue(), input.getFlowsPerDpn().intValue(),
257 input.getBatchSize().intValue(), input.getSleepFor().intValue(),
258 input.getSleepAfter().intValue(), input.getStartTableId().shortValue(),
259 input.getEndTableId().shortValue(), input.isCreateParents());
261 flowTester.deleteFlows(input.getDpnCount().intValue(), input.getFlowsPerDpn().intValue(),
262 input.getBatchSize().intValue(), input.getStartTableId().shortValue(),
263 input.getEndTableId().shortValue());
265 RpcResultBuilder<FlowTestOutput> rpcResultBuilder = RpcResultBuilder.success();
266 return Futures.immediateFuture(rpcResultBuilder.build());
269 FlowWriterSequential flowTester = new FlowWriterSequential(dataBroker, fjService);
270 flowCounterBeanImpl.setWriter(flowTester);
271 if (input.isIsAdd()) {
272 flowTester.addFlows(input.getDpnCount().intValue(), input.getFlowsPerDpn().intValue(),
273 input.getBatchSize().intValue(), input.getSleepFor().intValue(),
274 input.getStartTableId().shortValue(), input.getEndTableId().shortValue(),
275 input.isCreateParents());
277 flowTester.deleteFlows(input.getDpnCount().intValue(), input.getFlowsPerDpn().intValue(),
278 input.getBatchSize().intValue(), input.getStartTableId().shortValue(),
279 input.getEndTableId().shortValue());
282 FlowWriterConcurrent flowTester = new FlowWriterConcurrent(dataBroker, fjService);
283 flowCounterBeanImpl.setWriter(flowTester);
284 if (input.isIsAdd()) {
285 flowTester.addFlows(input.getDpnCount().intValue(), input.getFlowsPerDpn().intValue(),
286 input.getBatchSize().intValue(), input.getSleepFor().intValue(),
287 input.getSleepAfter().intValue(), input.getStartTableId().shortValue(),
288 input.getEndTableId().shortValue(), input.isCreateParents());
290 flowTester.deleteFlows(input.getDpnCount().intValue(), input.getFlowsPerDpn().intValue(),
291 input.getBatchSize().intValue(), input.getStartTableId().shortValue(),
292 input.getEndTableId().shortValue());
295 RpcResultBuilder<FlowTestOutput> rpcResultBuilder = RpcResultBuilder.success();
296 return Futures.immediateFuture(rpcResultBuilder.build());
300 public ListenableFuture<RpcResult<TableTestOutput>> tableTest(final TableTestInput input) {
301 final TableWriter writer = new TableWriter(dataBroker, fjService);
302 flowCounterBeanImpl.setWriter(writer);
303 switch (input.getOperation()) {
305 writer.addTables(input.getDpnCount().intValue(), input.getStartTableId().shortValue(),
306 input.getEndTableId().shortValue());
309 writer.deleteTables(input.getDpnCount().intValue(), input.getStartTableId().shortValue(),
310 input.getEndTableId().shortValue());
313 RpcResultBuilder<TableTestOutput> rpcResultBuilder = RpcResultBuilder.failed();
314 return Futures.immediateFuture(rpcResultBuilder.build());
316 RpcResultBuilder<TableTestOutput> rpcResultBuilder = RpcResultBuilder.success();
317 return Futures.immediateFuture(rpcResultBuilder.build());
321 public ListenableFuture<RpcResult<FlowRpcAddMultipleOutput>> flowRpcAddMultiple(
322 final FlowRpcAddMultipleInput input) {
323 FlowWriterDirectOFRpc flowTesterRPC = new FlowWriterDirectOFRpc(dataBroker, flowService, fjService);
324 flowTesterRPC.rpcFlowAddAll(input.getFlowCount().intValue(), input.getRpcBatchSize().intValue());
325 RpcResultBuilder<FlowRpcAddMultipleOutput> rpcResultBuilder = RpcResultBuilder.success();
326 return Futures.immediateFuture(rpcResultBuilder.build());