gerrit.fd Code Review - hc2vpp.git/blob

   1 /*
   2  * Copyright (c) 2016 Cisco and/or its affiliates.
   3  *
   4  * Licensed under the Apache License, Version 2.0 (the "License");
   5  * you may not use this file except in compliance with the License.
   6  * You may obtain a copy of the License at:
   7  *
   8  *     http://www.apache.org/licenses/LICENSE-2.0
   9  *
  10  * Unless required by applicable law or agreed to in writing, software
  11  * distributed under the License is distributed on an "AS IS" BASIS,
  12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13  * See the License for the specific language governing permissions and
  14  * limitations under the License.
  15  */
  16
  17 package io.fd.honeycomb.v3po.translate.util.write.registry;
  18
  19 import static com.google.common.base.Preconditions.checkArgument;
  20
  21 import com.google.common.base.Optional;
  22 import com.google.common.collect.HashMultimap;
  23 import com.google.common.collect.ImmutableMap;
  24 import com.google.common.collect.ImmutableMultimap;
  25 import com.google.common.collect.ImmutableSet;
  26 import com.google.common.collect.Lists;
  27 import com.google.common.collect.Multimap;
  28 import com.google.common.collect.Sets;
  29 import io.fd.honeycomb.v3po.translate.TranslationException;
  30 import io.fd.honeycomb.v3po.translate.util.RWUtils;
  31 import io.fd.honeycomb.v3po.translate.write.DataObjectUpdate;
  32 import io.fd.honeycomb.v3po.translate.write.WriteContext;
  33 import io.fd.honeycomb.v3po.translate.write.WriteFailedException;
  34 import io.fd.honeycomb.v3po.translate.write.Writer;
  35 import io.fd.honeycomb.v3po.translate.write.WriterRegistry;
  36 import java.util.Collection;
  37 import java.util.Collections;
  38 import java.util.HashSet;
  39 import java.util.Map;
  40 import java.util.Set;
  41 import java.util.stream.Collectors;
  42 import javax.annotation.Nonnull;
  43 import javax.annotation.Nullable;
  44 import javax.annotation.concurrent.ThreadSafe;
  45 import org.opendaylight.yangtools.yang.binding.DataObject;
  46 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  47 import org.slf4j.Logger;
  48 import org.slf4j.LoggerFactory;
  49
  50 /**
  51  * Flat writer registry, delegating updates to writers in the order writers were submitted.
  52  */
  53 @ThreadSafe
  54 final class FlatWriterRegistry implements WriterRegistry {
  55
  56     private static final Logger LOG = LoggerFactory.getLogger(FlatWriterRegistry.class);
  57
  58     // All types handled by writers directly or as children
  59     private final ImmutableSet<InstanceIdentifier<?>> handledTypes;
  60
  61     private final Set<InstanceIdentifier<?>> writersOrderReversed;
  62     private final Set<InstanceIdentifier<?>> writersOrder;
  63     private final Map<InstanceIdentifier<?>, Writer<?>> writers;
  64
  65     /**
  66      * Create flat registry instance.
  67      *
  68      * @param writers immutable, ordered map of writers to use to process updates. Order of the writers has to be
  69      *                one in which create and update operations should be handled. Deletes will be handled in reversed
  70      *                order. All deletes are handled before handling all the updates.
  71      */
  72     FlatWriterRegistry(@Nonnull final ImmutableMap<InstanceIdentifier<?>, Writer<?>> writers) {
  73         this.writers = writers;
  74         this.writersOrderReversed = Sets.newLinkedHashSet(Lists.reverse(Lists.newArrayList(writers.keySet())));
  75         this.writersOrder = writers.keySet();
  76         this.handledTypes = getAllHandledTypes(writers);
  77     }
  78
  79     private static ImmutableSet<InstanceIdentifier<?>> getAllHandledTypes(
  80             @Nonnull final ImmutableMap<InstanceIdentifier<?>, Writer<?>> writers) {
  81         final ImmutableSet.Builder<InstanceIdentifier<?>> handledTypesBuilder = ImmutableSet.builder();
  82         for (Map.Entry<InstanceIdentifier<?>, Writer<?>> writerEntry : writers.entrySet()) {
  83             final InstanceIdentifier<?> writerType = writerEntry.getKey();
  84             final Writer<?> writer = writerEntry.getValue();
  85             handledTypesBuilder.add(writerType);
  86             if (writer instanceof SubtreeWriter) {
  87                 handledTypesBuilder.addAll(((SubtreeWriter<?>) writer).getHandledChildTypes());
  88             }
  89         }
  90         return handledTypesBuilder.build();
  91     }
  92
  93     @Override
  94     public void update(@Nonnull final InstanceIdentifier<? extends DataObject> id,
  95                        @Nullable final DataObject dataBefore,
  96                        @Nullable final DataObject dataAfter,
  97                        @Nonnull final WriteContext ctx) throws WriteFailedException {
  98         singleUpdate(ImmutableMultimap.of(
  99                 RWUtils.makeIidWildcarded(id), DataObjectUpdate.create(id, dataBefore, dataAfter)), ctx);
 100     }
 101
 102     @Override
 103     public void update(@Nonnull final DataObjectUpdates updates,
 104                        @Nonnull final WriteContext ctx) throws TranslationException {
 105         if (updates.isEmpty()) {
 106             return;
 107         }
 108
 109         // Optimization
 110         if (updates.containsOnlySingleType()) {
 111             // First process delete
 112             singleUpdate(updates.getDeletes(), ctx);
 113             // Next is update
 114             singleUpdate(updates.getUpdates(), ctx);
 115         } else {
 116             // First process deletes
 117             bulkUpdate(updates.getDeletes(), ctx, true, writersOrderReversed);
 118             // Next are updates
 119             bulkUpdate(updates.getUpdates(), ctx, true, writersOrder);
 120         }
 121
 122         LOG.debug("Update successful for types: {}", updates.getTypeIntersection());
 123         LOG.trace("Update successful for: {}", updates);
 124     }
 125
 126     private void singleUpdate(@Nonnull final Multimap<InstanceIdentifier<?>, ? extends DataObjectUpdate> updates,
 127                               @Nonnull final WriteContext ctx) throws WriteFailedException {
 128         if (updates.isEmpty()) {
 129             return;
 130         }
 131
 132         final InstanceIdentifier<?> singleType = updates.keySet().iterator().next();
 133         LOG.debug("Performing single type update for: {}", singleType);
 134         Collection<? extends DataObjectUpdate> singleTypeUpdates = updates.get(singleType);
 135         Writer<?> writer = getWriter(singleType);
 136
 137         if (writer == null) {
 138             // This node must be handled by a subtree writer, find it and call it or else fail
 139             checkArgument(handledTypes.contains(singleType), "Unable to process update. Missing writers for: %s",
 140                     singleType);
 141             writer = getSubtreeWriterResponsible(singleType);
 142             singleTypeUpdates = getParentDataObjectUpdate(ctx, updates, writer);
 143         }
 144
 145         LOG.trace("Performing single type update with writer: {}", writer);
 146         for (DataObjectUpdate singleUpdate : singleTypeUpdates) {
 147             writer.update(singleUpdate.getId(), singleUpdate.getDataBefore(), singleUpdate.getDataAfter(), ctx);
 148         }
 149     }
 150
 151     private Writer<?> getSubtreeWriterResponsible(final InstanceIdentifier<?> singleType) {
 152         final Writer<?> writer;// This is slow ( minor TODO-perf )
 153         writer = writers.values().stream()
 154                 .filter(w -> w instanceof SubtreeWriter)
 155                 .filter(w -> ((SubtreeWriter<?>) w).getHandledChildTypes().contains(singleType))
 156                 .findFirst()
 157                 .get();
 158         return writer;
 159     }
 160
 161     private Collection<DataObjectUpdate> getParentDataObjectUpdate(final WriteContext ctx,
 162                                                                    final Multimap<InstanceIdentifier<?>, ? extends DataObjectUpdate> updates,
 163                                                                    final Writer<?> writer) {
 164         // Now read data for subtree reader root, but first keyed ID is needed and that ID can be cut from updates
 165         InstanceIdentifier<?> firstAffectedChildId = ((SubtreeWriter<?>) writer).getHandledChildTypes().stream()
 166                 .filter(updates::containsKey)
 167                 .map(unkeyedId -> updates.get(unkeyedId))
 168                 .flatMap(doUpdates -> doUpdates.stream())
 169                 .map(DataObjectUpdate::getId)
 170                 .findFirst()
 171                 .get();
 172
 173         final InstanceIdentifier<?> parentKeyedId =
 174                 RWUtils.cutId(firstAffectedChildId, writer.getManagedDataObjectType());
 175
 176         final Optional<? extends DataObject> parentBefore = ctx.readBefore(parentKeyedId);
 177         final Optional<? extends DataObject> parentAfter = ctx.readAfter(parentKeyedId);
 178         return Collections.singleton(
 179                 DataObjectUpdate.create(parentKeyedId, parentBefore.orNull(), parentAfter.orNull()));
 180     }
 181
 182     private void bulkUpdate(@Nonnull final Multimap<InstanceIdentifier<?>, ? extends DataObjectUpdate> updates,
 183                             @Nonnull final WriteContext ctx,
 184                             final boolean attemptRevert,
 185                             @Nonnull final Set<InstanceIdentifier<?>> writersOrder) throws BulkUpdateException {
 186         if (updates.isEmpty()) {
 187             return;
 188         }
 189
 190         LOG.debug("Performing bulk update with revert attempt: {} for: {}", attemptRevert, updates.keySet());
 191
 192         // Check that all updates can be handled
 193         checkAllTypesCanBeHandled(updates);
 194
 195         // Capture all changes successfully processed in case revert is needed
 196         final Set<InstanceIdentifier<?>> processedNodes = new HashSet<>();
 197
 198         // Iterate over all writers and call update if there are any related updates
 199         for (InstanceIdentifier<?> writerType : writersOrder) {
 200             Collection<? extends DataObjectUpdate> writersData = updates.get(writerType);
 201             final Writer<?> writer = getWriter(writerType);
 202
 203             if (writersData.isEmpty()) {
 204                 // If there are no data for current writer, but it is a SubtreeWriter and there are updates to
 205                 // its children, still invoke it with its root data
 206                 if (writer instanceof SubtreeWriter<?> && isAffected(((SubtreeWriter<?>) writer), updates)) {
 207                     // Provide parent data for SubtreeWriter for further processing
 208                     writersData = getParentDataObjectUpdate(ctx, updates, writer);
 209                 } else {
 210                     // Skipping unaffected writer
 211                     // Alternative to this would be modification sort according to the order of writers
 212                     continue;
 213                 }
 214             }
 215
 216             LOG.debug("Performing update for: {}",  writerType);
 217             LOG.trace("Performing update with writer: {}", writer);
 218
 219             for (DataObjectUpdate singleUpdate : writersData) {
 220                 try {
 221                     writer.update(singleUpdate.getId(), singleUpdate.getDataBefore(), singleUpdate.getDataAfter(), ctx);
 222                     processedNodes.add(singleUpdate.getId());
 223                     LOG.trace("Update successful for type: {}", writerType);
 224                     LOG.debug("Update successful for: {}", singleUpdate);
 225                 } catch (Exception e) {
 226                     LOG.error("Error while processing data change of: {} (updates={})", writerType, writersData, e);
 227
 228                     final Reverter reverter = attemptRevert
 229                             ? new ReverterImpl(processedNodes, updates, writersOrder, ctx)
 230                             : () -> {}; // NOOP reverter
 231
 232                     // Find out which changes left unprocessed
 233                     final Set<InstanceIdentifier<?>> unprocessedChanges = updates.values().stream()
 234                             .map(DataObjectUpdate::getId)
 235                             .filter(id -> !processedNodes.contains(id))
 236                             .collect(Collectors.toSet());
 237                     throw new BulkUpdateException(unprocessedChanges, reverter, e);
 238                 }
 239             }
 240         }
 241     }
 242
 243     private void checkAllTypesCanBeHandled(
 244             @Nonnull final Multimap<InstanceIdentifier<?>, ? extends DataObjectUpdate> updates) {
 245         if (!handledTypes.containsAll(updates.keySet())) {
 246             final Sets.SetView<InstanceIdentifier<?>> missingWriters = Sets.difference(updates.keySet(), handledTypes);
 247             LOG.warn("Unable to process update. Missing writers for: {}", missingWriters);
 248             throw new IllegalArgumentException("Unable to process update. Missing writers for: " + missingWriters);
 249         }
 250     }
 251
 252     /**
 253      * Check whether {@link SubtreeWriter} is affected by the updates.
 254      *
 255      * @return true if there are any updates to SubtreeWriter's child nodes (those marked by SubtreeWriter
 256      *         as being taken care of)
 257      * */
 258     private static boolean isAffected(final SubtreeWriter<?> writer,
 259                                final Multimap<InstanceIdentifier<?>, ? extends DataObjectUpdate> updates) {
 260         return !Sets.intersection(writer.getHandledChildTypes(), updates.keySet()).isEmpty();
 261     }
 262
 263     @Nullable
 264     private Writer<?> getWriter(@Nonnull final InstanceIdentifier<?> singleType) {
 265         return writers.get(singleType);
 266     }
 267
 268     @Nonnull
 269     @Override
 270     public InstanceIdentifier<DataObject> getManagedDataObjectType() {
 271         throw new UnsupportedOperationException("Registry has no managed type");
 272     }
 273
 274     // FIXME unit test
 275     private final class ReverterImpl implements Reverter {
 276
 277         private final Collection<InstanceIdentifier<?>> processedNodes;
 278         private final Multimap<InstanceIdentifier<?>, ? extends DataObjectUpdate> updates;
 279         private final Set<InstanceIdentifier<?>> revertDeleteOrder;
 280         private final WriteContext ctx;
 281
 282         ReverterImpl(final Collection<InstanceIdentifier<?>> processedNodes,
 283                      final Multimap<InstanceIdentifier<?>, ? extends DataObjectUpdate> updates,
 284                      final Set<InstanceIdentifier<?>> writersOrderOriginal,
 285                      final WriteContext ctx) {
 286             this.processedNodes = processedNodes;
 287             this.updates = updates;
 288             // Use opposite ordering when executing revert
 289             this.revertDeleteOrder =  writersOrderOriginal == FlatWriterRegistry.this.writersOrder
 290                     ? FlatWriterRegistry.this.writersOrderReversed
 291                     : FlatWriterRegistry.this.writersOrder;
 292             this.ctx = ctx;
 293         }
 294
 295         @Override
 296         public void revert() throws RevertFailedException {
 297             Multimap<InstanceIdentifier<?>, DataObjectUpdate> updatesToRevert =
 298                     filterAndRevertProcessed(updates, processedNodes);
 299
 300             LOG.info("Attempting revert for changes: {}", updatesToRevert);
 301             try {
 302                 // Perform reversed bulk update without revert attempt
 303                 bulkUpdate(updatesToRevert, ctx, true, revertDeleteOrder);
 304                 LOG.info("Revert successful");
 305             } catch (BulkUpdateException e) {
 306                 LOG.error("Revert failed", e);
 307                 throw new RevertFailedException(e.getFailedIds(), e);
 308             }
 309         }
 310
 311         /**
 312          * Create new updates map, but only keep already processed changes. Switching before and after data for each
 313          * update.
 314          */
 315         private Multimap<InstanceIdentifier<?>, DataObjectUpdate> filterAndRevertProcessed(
 316                 final Multimap<InstanceIdentifier<?>, ? extends DataObjectUpdate> updates,
 317                 final Collection<InstanceIdentifier<?>> processedNodes) {
 318             final Multimap<InstanceIdentifier<?>, DataObjectUpdate> filtered = HashMultimap.create();
 319             for (InstanceIdentifier<?> processedNode : processedNodes) {
 320                 final InstanceIdentifier<?> wildcardedIid = RWUtils.makeIidWildcarded(processedNode);
 321                 if (updates.containsKey(wildcardedIid)) {
 322                     updates.get(wildcardedIid).stream()
 323                             .filter(dataObjectUpdate -> processedNode.contains(dataObjectUpdate.getId()))
 324                             .forEach(dataObjectUpdate -> filtered.put(processedNode, dataObjectUpdate.reverse()));
 325                 }
 326             }
 327             return filtered;
 328         }
 329     }
 330
 331 }