opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/sharding/PrefixedShardConfigWriter.java

   1 /*
   2  * Copyright (c) 2017 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.controller.cluster.sharding;
  10
  11 import com.google.common.util.concurrent.AsyncFunction;
  12 import com.google.common.util.concurrent.Futures;
  13 import com.google.common.util.concurrent.ListenableFuture;
  14 import com.google.common.util.concurrent.MoreExecutors;
  15 import java.util.Collection;
  16 import java.util.concurrent.ExecutionException;
  17 import org.opendaylight.controller.cluster.access.concepts.MemberName;
  18 import org.opendaylight.controller.cluster.databroker.actors.dds.ClientLocalHistory;
  19 import org.opendaylight.controller.cluster.databroker.actors.dds.ClientSnapshot;
  20 import org.opendaylight.controller.cluster.databroker.actors.dds.ClientTransaction;
  21 import org.opendaylight.controller.cluster.databroker.actors.dds.DataStoreClient;
  22 import org.opendaylight.controller.cluster.datastore.utils.ClusterUtils;
  23 import org.opendaylight.mdsal.dom.api.DOMDataTreeWriteCursor;
  24 import org.opendaylight.mdsal.dom.spi.store.DOMStoreThreePhaseCommitCohort;
  25 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  26 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.NodeIdentifier;
  27 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.NodeIdentifierWithPredicates;
  28 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.NodeWithValue;
  29 import org.opendaylight.yangtools.yang.data.api.schema.ContainerNode;
  30 import org.opendaylight.yangtools.yang.data.api.schema.LeafSetEntryNode;
  31 import org.opendaylight.yangtools.yang.data.api.schema.MapEntryNode;
  32 import org.opendaylight.yangtools.yang.data.impl.schema.builder.api.ListNodeBuilder;
  33 import org.opendaylight.yangtools.yang.data.impl.schema.builder.impl.ImmutableContainerNodeBuilder;
  34 import org.opendaylight.yangtools.yang.data.impl.schema.builder.impl.ImmutableLeafNodeBuilder;
  35 import org.opendaylight.yangtools.yang.data.impl.schema.builder.impl.ImmutableLeafSetEntryNodeBuilder;
  36 import org.opendaylight.yangtools.yang.data.impl.schema.builder.impl.ImmutableLeafSetNodeBuilder;
  37 import org.opendaylight.yangtools.yang.data.impl.schema.builder.impl.ImmutableMapEntryNodeBuilder;
  38 import org.opendaylight.yangtools.yang.data.impl.schema.builder.impl.ImmutableMapNodeBuilder;
  39 import org.slf4j.Logger;
  40 import org.slf4j.LoggerFactory;
  41
  42 /**
  43  * Writes and removes prefix-based shards' configuration
  44  * to prefix-shard-configuration. This classed is meant to be utilized
  45  * by {@link DistributedShardedDOMDataTree} for updating
  46  * prefix-shard-configuration upon creating and de-spawning prefix-based shards.
  47  */
  48 @Deprecated(forRemoval = true)
  49 class PrefixedShardConfigWriter {
  50
  51     private static final Logger LOG = LoggerFactory.getLogger(PrefixedShardConfigWriter.class);
  52
  53     private final ClientLocalHistory history;
  54
  55     PrefixedShardConfigWriter(final DataStoreClient client) {
  56         history = client.createLocalHistory();
  57         writeInitialParent();
  58     }
  59
  60     ListenableFuture<Void> writeConfig(final YangInstanceIdentifier path, final Collection<MemberName> replicas) {
  61         LOG.debug("Writing config for {}, replicas {}", path, replicas);
  62
  63         return doSubmit(doWrite(path, replicas));
  64     }
  65
  66     ListenableFuture<Void> removeConfig(final YangInstanceIdentifier path) {
  67         LOG.debug("Removing config for {}.", path);
  68
  69         return doSubmit(doDelete(path));
  70     }
  71
  72     private void writeInitialParent() {
  73         final ClientTransaction tx = history.createTransaction();
  74
  75         final DOMDataTreeWriteCursor cursor = tx.openCursor();
  76
  77         final ContainerNode root = ImmutableContainerNodeBuilder.create()
  78                 .withNodeIdentifier(new NodeIdentifier(ClusterUtils.PREFIX_SHARDS_QNAME))
  79                 .withChild(ImmutableMapNodeBuilder.create()
  80                         .withNodeIdentifier(new NodeIdentifier(ClusterUtils.SHARD_LIST_QNAME))
  81                         .build())
  82                 .build();
  83
  84         cursor.merge(ClusterUtils.PREFIX_SHARDS_PATH.getLastPathArgument(), root);
  85         cursor.close();
  86
  87         final DOMStoreThreePhaseCommitCohort cohort = tx.ready();
  88
  89         submitBlocking(cohort);
  90     }
  91
  92     private static void submitBlocking(final DOMStoreThreePhaseCommitCohort cohort) {
  93         try {
  94             doSubmit(cohort).get();
  95         } catch (final InterruptedException | ExecutionException e) {
  96             LOG.error("Unable to write initial shard config parent.", e);
  97         }
  98     }
  99
 100     private static ListenableFuture<Void> doSubmit(final DOMStoreThreePhaseCommitCohort cohort) {
 101         final AsyncFunction<Boolean, Void> validateFunction = input -> cohort.preCommit();
 102         final AsyncFunction<Void, Void> prepareFunction = input -> cohort.commit();
 103
 104         final ListenableFuture<Void> prepareFuture = Futures.transformAsync(cohort.canCommit(), validateFunction,
 105             MoreExecutors.directExecutor());
 106         return Futures.transformAsync(prepareFuture, prepareFunction, MoreExecutors.directExecutor());
 107     }
 108
 109     boolean checkDefaultIsPresent() {
 110         final NodeIdentifierWithPredicates pag =
 111                 NodeIdentifierWithPredicates.of(ClusterUtils.SHARD_LIST_QNAME, ClusterUtils.SHARD_PREFIX_QNAME,
 112                 YangInstanceIdentifier.empty());
 113
 114         final YangInstanceIdentifier defaultId = ClusterUtils.SHARD_LIST_PATH.node(pag);
 115
 116         final ClientSnapshot snapshot = history.takeSnapshot();
 117         try {
 118             return snapshot.exists(defaultId).get();
 119         } catch (InterruptedException | ExecutionException e) {
 120             LOG.error("Presence check of default shard in configuration failed.", e);
 121             return false;
 122         } finally {
 123             snapshot.abort();
 124         }
 125     }
 126
 127     private DOMStoreThreePhaseCommitCohort doWrite(final YangInstanceIdentifier path,
 128                                                    final Collection<MemberName> replicas) {
 129
 130         final ListNodeBuilder<Object, LeafSetEntryNode<Object>> replicaListBuilder =
 131                 ImmutableLeafSetNodeBuilder.create().withNodeIdentifier(
 132                         new NodeIdentifier(ClusterUtils.SHARD_REPLICA_QNAME));
 133
 134         replicas.forEach(name -> replicaListBuilder.withChild(
 135                 ImmutableLeafSetEntryNodeBuilder.create()
 136                         .withNodeIdentifier(new NodeWithValue<>(ClusterUtils.SHARD_REPLICA_QNAME, name.getName()))
 137                         .withValue(name.getName())
 138                         .build()));
 139
 140         final MapEntryNode newEntry = ImmutableMapEntryNodeBuilder.create()
 141                 .withNodeIdentifier(
 142                         NodeIdentifierWithPredicates.of(ClusterUtils.SHARD_LIST_QNAME, ClusterUtils.SHARD_PREFIX_QNAME,
 143                                 path))
 144                 .withChild(ImmutableLeafNodeBuilder.create()
 145                         .withNodeIdentifier(new NodeIdentifier(ClusterUtils.SHARD_PREFIX_QNAME))
 146                         .withValue(path)
 147                         .build())
 148                 .withChild(ImmutableContainerNodeBuilder.create()
 149                         .withNodeIdentifier(new NodeIdentifier(ClusterUtils.SHARD_REPLICAS_QNAME))
 150                         .withChild(replicaListBuilder.build())
 151                         .build())
 152                 .build();
 153
 154         final ClientTransaction tx = history.createTransaction();
 155         final DOMDataTreeWriteCursor cursor = tx.openCursor();
 156
 157         ClusterUtils.SHARD_LIST_PATH.getPathArguments().forEach(cursor::enter);
 158
 159         cursor.write(newEntry.getIdentifier(), newEntry);
 160         cursor.close();
 161
 162         return tx.ready();
 163     }
 164
 165     private DOMStoreThreePhaseCommitCohort doDelete(final YangInstanceIdentifier path) {
 166
 167         final ClientTransaction tx = history.createTransaction();
 168         final DOMDataTreeWriteCursor cursor = tx.openCursor();
 169
 170         ClusterUtils.SHARD_LIST_PATH.getPathArguments().forEach(cursor::enter);
 171
 172         cursor.delete(
 173                 NodeIdentifierWithPredicates.of(ClusterUtils.SHARD_LIST_QNAME, ClusterUtils.SHARD_PREFIX_QNAME, path));
 174         cursor.close();
 175
 176         return tx.ready();
 177     }
 178 }