2 * Copyright (c) 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
9 package org.opendaylight.controller.cluster.sharding;
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.common.api.ReadFailedException;
24 import org.opendaylight.mdsal.dom.api.DOMDataTreeWriteCursor;
25 import org.opendaylight.mdsal.dom.spi.store.DOMStoreThreePhaseCommitCohort;
26 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
27 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.NodeIdentifier;
28 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.NodeIdentifierWithPredicates;
29 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.NodeWithValue;
30 import org.opendaylight.yangtools.yang.data.api.schema.ContainerNode;
31 import org.opendaylight.yangtools.yang.data.api.schema.LeafSetEntryNode;
32 import org.opendaylight.yangtools.yang.data.api.schema.MapEntryNode;
33 import org.opendaylight.yangtools.yang.data.impl.schema.builder.api.ListNodeBuilder;
34 import org.opendaylight.yangtools.yang.data.impl.schema.builder.impl.ImmutableContainerNodeBuilder;
35 import org.opendaylight.yangtools.yang.data.impl.schema.builder.impl.ImmutableLeafNodeBuilder;
36 import org.opendaylight.yangtools.yang.data.impl.schema.builder.impl.ImmutableLeafSetEntryNodeBuilder;
37 import org.opendaylight.yangtools.yang.data.impl.schema.builder.impl.ImmutableLeafSetNodeBuilder;
38 import org.opendaylight.yangtools.yang.data.impl.schema.builder.impl.ImmutableMapEntryNodeBuilder;
39 import org.opendaylight.yangtools.yang.data.impl.schema.builder.impl.ImmutableMapNodeBuilder;
40 import org.slf4j.Logger;
41 import org.slf4j.LoggerFactory;
44 * Writes and removes prefix-based shards' configuration
45 * to prefix-shard-configuration. This classed is meant to be utilized
46 * by {@link DistributedShardedDOMDataTree} for updating
47 * prefix-shard-configuration upon creating and de-spawning prefix-based shards.
49 class PrefixedShardConfigWriter {
51 private static final Logger LOG = LoggerFactory.getLogger(PrefixedShardConfigWriter.class);
53 private final ClientLocalHistory history;
55 PrefixedShardConfigWriter(final DataStoreClient client) {
56 history = client.createLocalHistory();
60 ListenableFuture<Void> writeConfig(final YangInstanceIdentifier path, final Collection<MemberName> replicas) {
61 LOG.debug("Writing config for {}, replicas {}", path, replicas);
63 return doSubmit(doWrite(path, replicas));
66 ListenableFuture<Void> removeConfig(final YangInstanceIdentifier path) {
67 LOG.debug("Removing config for {}.", path);
69 return doSubmit(doDelete(path));
72 private void writeInitialParent() {
73 final ClientTransaction tx = history.createTransaction();
75 final DOMDataTreeWriteCursor cursor = tx.openCursor();
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))
84 cursor.merge(ClusterUtils.PREFIX_SHARDS_PATH.getLastPathArgument(), root);
87 final DOMStoreThreePhaseCommitCohort cohort = tx.ready();
89 submitBlocking(cohort);
92 private static void submitBlocking(final DOMStoreThreePhaseCommitCohort cohort) {
94 doSubmit(cohort).get();
95 } catch (final InterruptedException | ExecutionException e) {
96 LOG.error("Unable to write initial shard config parent.", e);
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();
104 final ListenableFuture<Void> prepareFuture = Futures.transformAsync(cohort.canCommit(), validateFunction,
105 MoreExecutors.directExecutor());
106 return Futures.transformAsync(prepareFuture, prepareFunction, MoreExecutors.directExecutor());
109 boolean checkDefaultIsPresent() {
110 final NodeIdentifierWithPredicates pag =
111 new NodeIdentifierWithPredicates(ClusterUtils.SHARD_LIST_QNAME, ClusterUtils.SHARD_PREFIX_QNAME,
112 YangInstanceIdentifier.EMPTY);
114 final YangInstanceIdentifier defaultId = ClusterUtils.SHARD_LIST_PATH.node(pag);
116 final ClientSnapshot snapshot = history.takeSnapshot();
118 return snapshot.exists(defaultId).checkedGet();
119 } catch (final ReadFailedException e) {
120 LOG.error("Presence check of default shard in configuration failed.", e);
127 private DOMStoreThreePhaseCommitCohort doWrite(final YangInstanceIdentifier path,
128 final Collection<MemberName> replicas) {
130 final ListNodeBuilder<Object, LeafSetEntryNode<Object>> replicaListBuilder =
131 ImmutableLeafSetNodeBuilder.create().withNodeIdentifier(
132 new NodeIdentifier(ClusterUtils.SHARD_REPLICA_QNAME));
134 replicas.forEach(name -> replicaListBuilder.withChild(
135 ImmutableLeafSetEntryNodeBuilder.create()
136 .withNodeIdentifier(new NodeWithValue<>(ClusterUtils.SHARD_REPLICA_QNAME, name.getName()))
137 .withValue(name.getName())
140 final MapEntryNode newEntry = ImmutableMapEntryNodeBuilder.create()
142 new NodeIdentifierWithPredicates(ClusterUtils.SHARD_LIST_QNAME, ClusterUtils.SHARD_PREFIX_QNAME,
144 .withChild(ImmutableLeafNodeBuilder.create()
145 .withNodeIdentifier(new NodeIdentifier(ClusterUtils.SHARD_PREFIX_QNAME))
148 .withChild(ImmutableContainerNodeBuilder.create()
149 .withNodeIdentifier(new NodeIdentifier(ClusterUtils.SHARD_REPLICAS_QNAME))
150 .withChild(replicaListBuilder.build())
154 final ClientTransaction tx = history.createTransaction();
155 final DOMDataTreeWriteCursor cursor = tx.openCursor();
157 ClusterUtils.SHARD_LIST_PATH.getPathArguments().forEach(cursor::enter);
159 cursor.write(newEntry.getIdentifier(), newEntry);
165 private DOMStoreThreePhaseCommitCohort doDelete(final YangInstanceIdentifier path) {
167 final ClientTransaction tx = history.createTransaction();
168 final DOMDataTreeWriteCursor cursor = tx.openCursor();
170 ClusterUtils.SHARD_LIST_PATH.getPathArguments().forEach(cursor::enter);
173 new NodeIdentifierWithPredicates(ClusterUtils.SHARD_LIST_QNAME, ClusterUtils.SHARD_PREFIX_QNAME, path));