openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/statistics/StatisticsGatheringUtils.java

   1 /*
   2  * Copyright (c) 2015 Cisco Systems, Inc. and others.  All rights reserved.
   3  *
   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
   7  */
   8
   9 package org.opendaylight.openflowplugin.impl.statistics;
  10
  11 import com.google.common.base.Function;
  12 import com.google.common.base.Optional;
  13 import com.google.common.util.concurrent.CheckedFuture;
  14 import com.google.common.util.concurrent.Futures;
  15 import com.google.common.util.concurrent.ListenableFuture;
  16 import com.google.common.util.concurrent.ListeningExecutorService;
  17 import com.google.common.util.concurrent.MoreExecutors;
  18 import java.text.SimpleDateFormat;
  19 import java.util.Collections;
  20 import java.util.Date;
  21 import java.util.List;
  22 import java.util.Objects;
  23 import java.util.concurrent.ExecutionException;
  24 import java.util.concurrent.atomic.AtomicBoolean;
  25 import java.util.stream.Collectors;
  26 import org.opendaylight.controller.md.sal.binding.api.ReadOnlyTransaction;
  27 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
  28 import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException;
  29 import org.opendaylight.controller.md.sal.common.api.data.TransactionChainClosedException;
  30 import org.opendaylight.openflowplugin.api.openflow.device.DeviceInfo;
  31 import org.opendaylight.openflowplugin.api.openflow.device.DeviceRegistry;
  32 import org.opendaylight.openflowplugin.api.openflow.device.TxFacade;
  33 import org.opendaylight.openflowplugin.api.openflow.registry.flow.DeviceFlowRegistry;
  34 import org.opendaylight.openflowplugin.api.openflow.registry.group.DeviceGroupRegistry;
  35 import org.opendaylight.openflowplugin.api.openflow.registry.meter.DeviceMeterRegistry;
  36 import org.opendaylight.openflowplugin.api.openflow.statistics.ofpspecific.EventIdentifier;
  37 import org.opendaylight.openflowplugin.api.openflow.statistics.ofpspecific.StatisticsGatherer;
  38 import org.opendaylight.openflowplugin.impl.common.MultipartReplyTranslatorUtil;
  39 import org.opendaylight.openflowplugin.impl.datastore.MultipartWriterProvider;
  40 import org.opendaylight.openflowplugin.openflow.md.core.sal.convertor.ConvertorExecutor;
  41 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.yang.types.rev130715.DateAndTime;
  42 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode;
  43 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableStatisticsGatheringStatus;
  44 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableStatisticsGatheringStatusBuilder;
  45 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.meters.Meter;
  46 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.meters.MeterKey;
  47 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.snapshot.gathering.status.grouping.SnapshotGatheringStatusEnd;
  48 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.snapshot.gathering.status.grouping.SnapshotGatheringStatusEndBuilder;
  49 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.snapshot.gathering.status.grouping.SnapshotGatheringStatusStartBuilder;
  50 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.Table;
  51 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.TableBuilder;
  52 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.groups.Group;
  53 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.groups.GroupKey;
  54 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.common.types.rev130731.MultipartType;
  55 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.OfHeader;
  56 import org.opendaylight.yangtools.yang.binding.DataContainer;
  57 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  58 import org.slf4j.Logger;
  59 import org.slf4j.LoggerFactory;
  60
  61 /**
  62  * Utils for gathering statistics.
  63  */
  64 public final class StatisticsGatheringUtils {
  65
  66     private static final String DATE_AND_TIME_FORMAT = "yyyy-MM-dd'T'HH:mm:ss.SSSXXX";
  67     private static final Logger LOG = LoggerFactory.getLogger(StatisticsGatheringUtils.class);
  68     private static final String QUEUE2_REQCTX = "QUEUE2REQCTX-";
  69
  70     private StatisticsGatheringUtils() {
  71         throw new IllegalStateException("This class should not be instantiated.");
  72     }
  73
  74     static <T extends OfHeader> ListenableFuture<Boolean> gatherStatistics(
  75             final StatisticsGatherer<T> statisticsGatheringService, final DeviceInfo deviceInfo,
  76             final MultipartType type, final TxFacade txFacade, final DeviceRegistry registry,
  77             final ConvertorExecutor convertorExecutor, final MultipartWriterProvider statisticsWriterProvider,
  78             final ListeningExecutorService executorService) {
  79         return Futures.transformAsync(statisticsGatheringService.getStatisticsOfType(
  80            new EventIdentifier(QUEUE2_REQCTX + type.toString(), deviceInfo.getNodeId().toString()), type),
  81             rpcResult -> executorService.submit(() -> {
  82                 final boolean rpcResultIsNull = rpcResult == null;
  83
  84                 if (!rpcResultIsNull && rpcResult.isSuccessful()) {
  85                     LOG.debug("Stats reply successfully received for node {} of type {}", deviceInfo.getNodeId(), type);
  86                     // TODO: in case the result value is null then multipart data probably got processed
  87                     // TODO: on the fly. This contract should by clearly stated and enforced.
  88                     // TODO: Now simple true value is returned
  89                     if (Objects.nonNull(rpcResult.getResult()) && !rpcResult.getResult().isEmpty()) {
  90                         final List<DataContainer> allMultipartData = rpcResult.getResult().stream()
  91                                 .map(reply -> MultipartReplyTranslatorUtil
  92                                         .translate(reply, deviceInfo, convertorExecutor, null))
  93                                 .filter(java.util.Optional::isPresent).map(java.util.Optional::get)
  94                                 .collect(Collectors.toList());
  95
  96                         return processStatistics(type, allMultipartData, txFacade, registry, deviceInfo,
  97                                                  statisticsWriterProvider);
  98                     } else {
  99                         LOG.debug("Stats reply was empty for node {} of type {}", deviceInfo.getNodeId(), type);
 100                     }
 101                 } else {
 102                     LOG.warn("Stats reply FAILED for node {} of type {}: {}", deviceInfo.getNodeId(), type,
 103                              rpcResultIsNull ? "" : rpcResult.getErrors());
 104                 }
 105                 return false;
 106             }), MoreExecutors.directExecutor());
 107     }
 108
 109     private static boolean processStatistics(final MultipartType type, final List<? extends DataContainer> statistics,
 110                                              final TxFacade txFacade, final DeviceRegistry deviceRegistry,
 111                                              final DeviceInfo deviceInfo,
 112                                              final MultipartWriterProvider statisticsWriterProvider) {
 113         final InstanceIdentifier<FlowCapableNode> instanceIdentifier = deviceInfo.getNodeInstanceIdentifier()
 114                 .augmentation(FlowCapableNode.class);
 115
 116         switch (type) {
 117             case OFPMPFLOW:
 118                 deleteAllKnownFlows(txFacade, instanceIdentifier, deviceRegistry.getDeviceFlowRegistry());
 119                 break;
 120             case OFPMPMETERCONFIG:
 121                 deleteAllKnownMeters(txFacade, instanceIdentifier, deviceRegistry.getDeviceMeterRegistry());
 122                 break;
 123             case OFPMPGROUPDESC:
 124                 deleteAllKnownGroups(txFacade, instanceIdentifier, deviceRegistry.getDeviceGroupRegistry());
 125                 break;
 126             default:
 127                 // no operation
 128         }
 129
 130         if (writeStatistics(type, statistics, deviceInfo, statisticsWriterProvider)) {
 131             txFacade.submitTransaction();
 132
 133             switch (type) {
 134                 case OFPMPFLOW:
 135                     deviceRegistry.getDeviceFlowRegistry().processMarks();
 136                     break;
 137                 case OFPMPMETERCONFIG:
 138                     deviceRegistry.getDeviceMeterRegistry().processMarks();
 139                     break;
 140                 case OFPMPGROUPDESC:
 141                     deviceRegistry.getDeviceGroupRegistry().processMarks();
 142                     break;
 143                 default:
 144                     // no operation
 145             }
 146
 147             LOG.debug("Stats reply added to transaction for node {} of type {}", deviceInfo.getNodeId(), type);
 148             return true;
 149         }
 150
 151         LOG.warn("Stats processing of type {} for node {} " + "failed during write-to-tx step", type, deviceInfo);
 152         return false;
 153     }
 154
 155     @SuppressWarnings("checkstyle:IllegalCatch")
 156     private static boolean writeStatistics(final MultipartType type, final List<? extends DataContainer> statistics,
 157                                            final DeviceInfo deviceInfo,
 158                                            final MultipartWriterProvider statisticsWriterProvider) {
 159         final AtomicBoolean result = new AtomicBoolean(false);
 160
 161         try {
 162             statistics.forEach(stat -> statisticsWriterProvider.lookup(type).ifPresent(p -> {
 163                 final boolean write = p.write(stat, false);
 164
 165                 if (!result.get()) {
 166                     result.set(write);
 167                 }
 168             }));
 169         } catch (final Exception ex) {
 170             LOG.warn("Stats processing of type {} for node {} " + "failed during write-to-tx step", type, deviceInfo,
 171                      ex);
 172         }
 173
 174         return result.get();
 175     }
 176
 177     public static void deleteAllKnownFlows(final TxFacade txFacade,
 178                                            final InstanceIdentifier<FlowCapableNode> instanceIdentifier,
 179                                            final DeviceFlowRegistry deviceFlowRegistry) {
 180         if (!txFacade.isTransactionsEnabled()) {
 181             return;
 182         }
 183
 184         final CheckedFuture<Optional<FlowCapableNode>, ReadFailedException> future;
 185         try (ReadOnlyTransaction readTx = txFacade.getReadTransaction()) {
 186             future = readTx.read(LogicalDatastoreType.OPERATIONAL, instanceIdentifier);
 187         }
 188
 189         try {
 190             Futures.transform(Futures.catchingAsync(future, Throwable.class, throwable -> {
 191                 return Futures.immediateFailedFuture(throwable);
 192             }, MoreExecutors.directExecutor()), (Function<Optional<FlowCapableNode>, Void>) flowCapNodeOpt -> {
 193                     // we have to read actual tables with all information before we set empty Flow list,
 194                     // merge is expensive and not applicable for lists
 195                     if (flowCapNodeOpt != null && flowCapNodeOpt.isPresent()) {
 196                         for (final Table tableData : flowCapNodeOpt.get().getTable()) {
 197                             final Table table = new TableBuilder(tableData).setFlow(Collections.emptyList()).build();
 198                             final InstanceIdentifier<Table> iiToTable = instanceIdentifier
 199                                     .child(Table.class, tableData.getKey());
 200                             txFacade.writeToTransaction(LogicalDatastoreType.OPERATIONAL, iiToTable, table);
 201                         }
 202                     }
 203                     return null;
 204                 }, MoreExecutors.directExecutor()).get();
 205         } catch (InterruptedException | ExecutionException ex) {
 206             LOG.debug("Failed to delete {} flows, exception: {}", deviceFlowRegistry.size(), ex);
 207         }
 208     }
 209
 210     public static void deleteAllKnownMeters(final TxFacade txFacade,
 211                                             final InstanceIdentifier<FlowCapableNode> instanceIdentifier,
 212                                             final DeviceMeterRegistry meterRegistry) {
 213         meterRegistry.forEach(meterId -> txFacade.addDeleteToTxChain(LogicalDatastoreType.OPERATIONAL,
 214                                                                      instanceIdentifier.child(Meter.class,
 215                                                                                               new MeterKey(meterId))));
 216     }
 217
 218     public static void deleteAllKnownGroups(final TxFacade txFacade,
 219                                             final InstanceIdentifier<FlowCapableNode> instanceIdentifier,
 220                                             final DeviceGroupRegistry groupRegistry) {
 221         groupRegistry.forEach(groupId -> txFacade.addDeleteToTxChain(LogicalDatastoreType.OPERATIONAL,
 222                                                                      instanceIdentifier.child(Group.class,
 223                                                                                               new GroupKey(groupId))));
 224     }
 225
 226     /**
 227      * Writes snapshot gathering start timestamp + cleans end mark.
 228      *
 229      * @param deviceInfo device info
 230      * @param txFacade   tx manager
 231      */
 232     static void markDeviceStateSnapshotStart(final DeviceInfo deviceInfo, final TxFacade txFacade) {
 233         final InstanceIdentifier<FlowCapableStatisticsGatheringStatus> statusPath = deviceInfo
 234                 .getNodeInstanceIdentifier().augmentation(FlowCapableStatisticsGatheringStatus.class);
 235
 236         final SimpleDateFormat simpleDateFormat = new SimpleDateFormat(DATE_AND_TIME_FORMAT);
 237         final FlowCapableStatisticsGatheringStatus gatheringStatus = new FlowCapableStatisticsGatheringStatusBuilder()
 238                 .setSnapshotGatheringStatusStart(new SnapshotGatheringStatusStartBuilder()
 239                                                          .setBegin(new DateAndTime(simpleDateFormat.format(new Date())))
 240                                                          .build())
 241                 .setSnapshotGatheringStatusEnd(null) // TODO: reconsider if really need to clean end mark here
 242                 .build();
 243         try {
 244             txFacade.writeToTransaction(LogicalDatastoreType.OPERATIONAL, statusPath, gatheringStatus);
 245         } catch (final TransactionChainClosedException e) {
 246             LOG.warn("Can't write to transaction, transaction chain probably closed.");
 247             LOG.trace("Write to transaction exception: ", e);
 248         }
 249
 250         txFacade.submitTransaction();
 251     }
 252
 253     /**
 254      * Writes snapshot gathering end timestamp + outcome.
 255      *
 256      * @param deviceInfo device info
 257      * @param txFacade   tx manager
 258      * @param succeeded  outcome of currently finished gathering
 259      */
 260     static void markDeviceStateSnapshotEnd(final DeviceInfo deviceInfo, final TxFacade txFacade,
 261                                            final boolean succeeded) {
 262         final InstanceIdentifier<SnapshotGatheringStatusEnd> statusEndPath = deviceInfo.getNodeInstanceIdentifier()
 263                 .augmentation(FlowCapableStatisticsGatheringStatus.class).child(SnapshotGatheringStatusEnd.class);
 264
 265         final SimpleDateFormat simpleDateFormat = new SimpleDateFormat(DATE_AND_TIME_FORMAT);
 266         final SnapshotGatheringStatusEnd gatheringStatus = new SnapshotGatheringStatusEndBuilder()
 267                 .setEnd(new DateAndTime(simpleDateFormat.format(new Date()))).setSucceeded(succeeded).build();
 268         try {
 269             txFacade.writeToTransaction(LogicalDatastoreType.OPERATIONAL, statusEndPath, gatheringStatus);
 270         } catch (TransactionChainClosedException e) {
 271             LOG.warn("Can't write to transaction, transaction chain probably closed.");
 272             LOG.trace("Write to transaction exception: ", e);
 273         }
 274
 275         txFacade.submitTransaction();
 276     }
 277 }