Deprecate DOMDataTreeProducer-related classes
[controller.git] / opendaylight / md-sal / sal-distributed-datastore / src / main / java / org / opendaylight / controller / cluster / sharding / DistributedShardChangePublisher.java
1 /*
2  * Copyright (c) 2016 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 package org.opendaylight.controller.cluster.sharding;
9
10 import static com.google.common.base.Preconditions.checkState;
11 import static java.util.Objects.requireNonNull;
12
13 import java.util.ArrayList;
14 import java.util.Collection;
15 import java.util.Collections;
16 import java.util.Iterator;
17 import java.util.LinkedList;
18 import java.util.List;
19 import java.util.Map;
20 import java.util.concurrent.ConcurrentHashMap;
21 import java.util.stream.Collectors;
22 import org.checkerframework.checker.lock.qual.GuardedBy;
23 import org.opendaylight.controller.cluster.databroker.actors.dds.DataStoreClient;
24 import org.opendaylight.controller.cluster.datastore.DistributedDataStoreInterface;
25 import org.opendaylight.mdsal.dom.api.DOMDataTreeChangeListener;
26 import org.opendaylight.mdsal.dom.api.DOMDataTreeIdentifier;
27 import org.opendaylight.mdsal.dom.spi.AbstractDOMDataTreeChangeListenerRegistration;
28 import org.opendaylight.mdsal.dom.spi.AbstractRegistrationTree;
29 import org.opendaylight.mdsal.dom.spi.RegistrationTreeNode;
30 import org.opendaylight.mdsal.dom.spi.shard.ChildShardContext;
31 import org.opendaylight.mdsal.dom.spi.store.DOMStoreTreeChangePublisher;
32 import org.opendaylight.yangtools.concepts.ListenerRegistration;
33 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
34 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.PathArgument;
35 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTree;
36 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidate;
37 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidateNode;
38 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidateNodes;
39 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidates;
40 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeConfiguration;
41 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeModification;
42 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataValidationFailedException;
43 import org.opendaylight.yangtools.yang.data.impl.schema.tree.InMemoryDataTreeFactory;
44 import org.opendaylight.yangtools.yang.data.impl.schema.tree.SchemaValidationFailedException;
45 import org.slf4j.Logger;
46 import org.slf4j.LoggerFactory;
47
48 @Deprecated(forRemoval = true)
49 public class DistributedShardChangePublisher
50         extends AbstractRegistrationTree<AbstractDOMDataTreeChangeListenerRegistration<?>>
51         implements DOMStoreTreeChangePublisher {
52
53     private static final Logger LOG = LoggerFactory.getLogger(DistributedShardChangePublisher.class);
54
55     private final DistributedDataStoreInterface distributedDataStore;
56     private final YangInstanceIdentifier shardPath;
57
58     private final Map<DOMDataTreeIdentifier, ChildShardContext> childShards;
59
60     @GuardedBy("this")
61     private final DataTree dataTree;
62
63     public DistributedShardChangePublisher(final DataStoreClient client,
64                                            final DistributedDataStoreInterface distributedDataStore,
65                                            final DOMDataTreeIdentifier prefix,
66                                            final Map<DOMDataTreeIdentifier, ChildShardContext> childShards) {
67         this.distributedDataStore = distributedDataStore;
68         // TODO keeping the whole dataTree thats contained in subshards doesn't seem like a good idea
69         // maybe the whole listener logic would be better in the backend shards where we have direct access to the
70         // dataTree and wont have to cache it redundantly.
71
72         final DataTreeConfiguration baseConfig;
73         switch (prefix.getDatastoreType()) {
74             case CONFIGURATION:
75                 baseConfig = DataTreeConfiguration.DEFAULT_CONFIGURATION;
76                 break;
77             case OPERATIONAL:
78                 baseConfig = DataTreeConfiguration.DEFAULT_OPERATIONAL;
79                 break;
80             default:
81                 throw new UnsupportedOperationException("Unknown prefix type " + prefix.getDatastoreType());
82         }
83
84         this.dataTree = new InMemoryDataTreeFactory().create(new DataTreeConfiguration.Builder(baseConfig.getTreeType())
85                 .setMandatoryNodesValidation(baseConfig.isMandatoryNodesValidationEnabled())
86                 .setUniqueIndexes(baseConfig.isUniqueIndexEnabled())
87                 .setRootPath(prefix.getRootIdentifier())
88                 .build());
89
90         // XXX: can we guarantee that the root is present in the schemacontext?
91         this.dataTree.setEffectiveModelContext(distributedDataStore.getActorUtils().getSchemaContext());
92         this.shardPath = prefix.getRootIdentifier();
93         this.childShards = childShards;
94     }
95
96     protected void registrationRemoved(final AbstractDOMDataTreeChangeListenerRegistration<?> registration) {
97         LOG.debug("Closing registration {}", registration);
98     }
99
100     @Override
101     public <L extends DOMDataTreeChangeListener> AbstractDOMDataTreeChangeListenerRegistration<L>
102             registerTreeChangeListener(final YangInstanceIdentifier path, final L listener) {
103         takeLock();
104         try {
105             return setupListenerContext(path, listener);
106         } finally {
107             releaseLock();
108         }
109     }
110
111     private <L extends DOMDataTreeChangeListener> AbstractDOMDataTreeChangeListenerRegistration<L>
112             setupListenerContext(final YangInstanceIdentifier listenerPath, final L listener) {
113         // we need to register the listener registration path based on the shards root
114         // we have to strip the shard path from the listener path and then register
115         YangInstanceIdentifier strippedIdentifier = listenerPath;
116         if (!shardPath.isEmpty()) {
117             strippedIdentifier = YangInstanceIdentifier.create(stripShardPath(shardPath, listenerPath));
118         }
119
120         final DOMDataTreeListenerWithSubshards subshardListener =
121                 new DOMDataTreeListenerWithSubshards(strippedIdentifier, listener);
122         final AbstractDOMDataTreeChangeListenerRegistration<L> reg =
123                 setupContextWithoutSubshards(listenerPath, strippedIdentifier, subshardListener);
124
125         for (final ChildShardContext maybeAffected : childShards.values()) {
126             if (listenerPath.contains(maybeAffected.getPrefix().getRootIdentifier())) {
127                 // consumer has initialDataChangeEvent subshard somewhere on lower level
128                 // register to the notification manager with snapshot and forward child notifications to parent
129                 LOG.debug("Adding new subshard{{}} to listener at {}", maybeAffected.getPrefix(), listenerPath);
130                 subshardListener.addSubshard(maybeAffected);
131             } else if (maybeAffected.getPrefix().getRootIdentifier().contains(listenerPath)) {
132                 // bind path is inside subshard
133                 // TODO can this happen? seems like in ShardedDOMDataTree we are
134                 // already registering to the lowest shard possible
135                 throw new UnsupportedOperationException("Listener should be registered directly "
136                         + "into initialDataChangeEvent subshard");
137             }
138         }
139
140         return reg;
141     }
142
143     private <L extends DOMDataTreeChangeListener> AbstractDOMDataTreeChangeListenerRegistration<L>
144             setupContextWithoutSubshards(final YangInstanceIdentifier shardLookup,
145                                          final YangInstanceIdentifier listenerPath,
146                                          final DOMDataTreeListenerWithSubshards listener) {
147
148         LOG.debug("Registering root listener full path: {}, path inside shard: {}", shardLookup, listenerPath);
149
150         // register in the shard tree
151         final RegistrationTreeNode<AbstractDOMDataTreeChangeListenerRegistration<?>> node =
152                 findNodeFor(listenerPath.getPathArguments());
153
154         // register listener in CDS
155         ListenerRegistration<DOMDataTreeChangeListener> listenerReg = distributedDataStore
156                 .registerProxyListener(shardLookup, listenerPath, listener);
157
158         @SuppressWarnings("unchecked")
159         final AbstractDOMDataTreeChangeListenerRegistration<L> registration =
160             new AbstractDOMDataTreeChangeListenerRegistration<>((L) listener) {
161                 @Override
162                 protected void removeRegistration() {
163                     listener.close();
164                     DistributedShardChangePublisher.this.removeRegistration(node, this);
165                     registrationRemoved(this);
166                     listenerReg.close();
167                 }
168             };
169         addRegistration(node, registration);
170
171         return registration;
172     }
173
174     private static Iterable<PathArgument> stripShardPath(final YangInstanceIdentifier shardPath,
175                                                          final YangInstanceIdentifier listenerPath) {
176         if (shardPath.isEmpty()) {
177             return listenerPath.getPathArguments();
178         }
179
180         final List<PathArgument> listenerPathArgs = new ArrayList<>(listenerPath.getPathArguments());
181         final Iterator<PathArgument> shardIter = shardPath.getPathArguments().iterator();
182         final Iterator<PathArgument> listenerIter = listenerPathArgs.iterator();
183
184         while (shardIter.hasNext()) {
185             if (shardIter.next().equals(listenerIter.next())) {
186                 listenerIter.remove();
187             } else {
188                 break;
189             }
190         }
191
192         return listenerPathArgs;
193     }
194
195     synchronized DataTreeCandidate applyChanges(final YangInstanceIdentifier listenerPath,
196             final Collection<DataTreeCandidate> changes) throws DataValidationFailedException {
197         final DataTreeModification modification = dataTree.takeSnapshot().newModification();
198         for (final DataTreeCandidate change : changes) {
199             try {
200                 DataTreeCandidates.applyToModification(modification, change);
201             } catch (SchemaValidationFailedException e) {
202                 LOG.error("Validation failed", e);
203             }
204         }
205
206         modification.ready();
207
208         final DataTreeCandidate candidate;
209
210         dataTree.validate(modification);
211
212         // strip nodes we dont need since this listener doesn't have to be registered at the root of the DataTree
213         candidate = dataTree.prepare(modification);
214         dataTree.commit(candidate);
215
216
217         DataTreeCandidateNode modifiedChild = candidate.getRootNode();
218
219         for (final PathArgument pathArgument : listenerPath.getPathArguments()) {
220             modifiedChild = modifiedChild.getModifiedChild(pathArgument).orElse(null);
221         }
222
223
224         if (modifiedChild == null) {
225             modifiedChild = DataTreeCandidateNodes.empty(dataTree.getRootPath().getLastPathArgument());
226         }
227
228         return DataTreeCandidates.newDataTreeCandidate(dataTree.getRootPath(), modifiedChild);
229     }
230
231
232     private final class DOMDataTreeListenerWithSubshards implements DOMDataTreeChangeListener {
233
234         private final YangInstanceIdentifier listenerPath;
235         private final DOMDataTreeChangeListener delegate;
236         private final Map<YangInstanceIdentifier, ListenerRegistration<DOMDataTreeChangeListener>> registrations =
237                 new ConcurrentHashMap<>();
238
239         @GuardedBy("this")
240         private final Collection<DataTreeCandidate> stashedDataTreeCandidates = new LinkedList<>();
241
242         DOMDataTreeListenerWithSubshards(final YangInstanceIdentifier listenerPath,
243                                          final DOMDataTreeChangeListener delegate) {
244             this.listenerPath = requireNonNull(listenerPath);
245             this.delegate = requireNonNull(delegate);
246         }
247
248         @Override
249         public synchronized void onDataTreeChanged(final Collection<DataTreeCandidate> changes) {
250             LOG.debug("Received data changed {}", changes);
251
252             if (!stashedDataTreeCandidates.isEmpty()) {
253                 LOG.debug("Adding stashed subshards' changes {}", stashedDataTreeCandidates);
254                 changes.addAll(stashedDataTreeCandidates);
255                 stashedDataTreeCandidates.clear();
256             }
257
258             try {
259                 applyChanges(listenerPath, changes);
260             } catch (final DataValidationFailedException e) {
261                 // TODO should we fail here? What if stashed changes
262                 // (changes from subshards) got ahead more than one generation
263                 // from current shard. Than we can fail to apply this changes
264                 // upon current data tree, but once we get respective changes
265                 // from current shard, we can apply also changes from
266                 // subshards.
267                 //
268                 // However, we can loose ability to notice and report some
269                 // errors then. For example, we cannot detect potential lost
270                 // changes from current shard.
271                 LOG.error("Validation failed for modification built from changes {}, current data tree: {}",
272                         changes, dataTree, e);
273                 throw new RuntimeException("Notification validation failed", e);
274             }
275
276             delegate.onDataTreeChanged(changes);
277         }
278
279         synchronized void onDataTreeChanged(final YangInstanceIdentifier pathFromRoot,
280                                             final Collection<DataTreeCandidate> changes) {
281             final YangInstanceIdentifier changeId =
282                     YangInstanceIdentifier.create(stripShardPath(dataTree.getRootPath(), pathFromRoot));
283
284             final List<DataTreeCandidate> newCandidates = changes.stream()
285                     .map(candidate -> DataTreeCandidates.newDataTreeCandidate(changeId, candidate.getRootNode()))
286                     .collect(Collectors.toList());
287
288             try {
289                 delegate.onDataTreeChanged(Collections.singleton(applyChanges(listenerPath, newCandidates)));
290             } catch (final DataValidationFailedException e) {
291                 // We cannot apply changes from subshard to current data tree.
292                 // Maybe changes from current shard haven't been applied to
293                 // data tree yet. Postpone processing of these changes till we
294                 // receive changes from current shard.
295                 LOG.debug("Validation for modification built from subshard {} changes {} failed, current data tree {}.",
296                         pathFromRoot, changes, dataTree, e);
297                 stashedDataTreeCandidates.addAll(newCandidates);
298             }
299         }
300
301         void addSubshard(final ChildShardContext context) {
302             checkState(context.getShard() instanceof DOMStoreTreeChangePublisher,
303                     "All subshards that are initialDataChangeEvent part of ListenerContext need to be listenable");
304
305             final DOMStoreTreeChangePublisher listenableShard = (DOMStoreTreeChangePublisher) context.getShard();
306             // since this is going into subshard we want to listen for ALL changes in the subshard
307             registrations.put(context.getPrefix().getRootIdentifier(),
308                     listenableShard.registerTreeChangeListener(
309                             context.getPrefix().getRootIdentifier(), changes -> onDataTreeChanged(
310                                     context.getPrefix().getRootIdentifier(), changes)));
311         }
312
313         void close() {
314             for (final ListenerRegistration<DOMDataTreeChangeListener> registration : registrations.values()) {
315                 registration.close();
316             }
317             registrations.clear();
318         }
319     }
320 }

©2013 OpenDaylight, A Linux Foundation Collaborative Project. All Rights Reserved.
OpenDaylight is a registered trademark of The OpenDaylight Project, Inc.
Linux Foundation and OpenDaylight are registered trademarks of the Linux Foundation.
Linux is a registered trademark of Linus Torvalds.