gerrit.opnfv Code Review - onosfw.git/blob

   1 /*
   2  * Copyright 2015 Open Networking Laboratory
   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 package org.onosproject.store.intent.impl;
  17
  18 import com.google.common.collect.ImmutableList;
  19
  20 import org.apache.commons.lang.math.RandomUtils;
  21 import org.apache.felix.scr.annotations.Activate;
  22 import org.apache.felix.scr.annotations.Component;
  23 import org.apache.felix.scr.annotations.Deactivate;
  24 import org.apache.felix.scr.annotations.Reference;
  25 import org.apache.felix.scr.annotations.ReferenceCardinality;
  26 import org.apache.felix.scr.annotations.Service;
  27 import org.onlab.util.KryoNamespace;
  28 import org.onosproject.cluster.ClusterService;
  29 import org.onosproject.cluster.ControllerNode;
  30 import org.onosproject.cluster.NodeId;
  31 import org.onosproject.net.intent.Intent;
  32 import org.onosproject.net.intent.IntentData;
  33 import org.onosproject.net.intent.IntentEvent;
  34 import org.onosproject.net.intent.IntentState;
  35 import org.onosproject.net.intent.IntentStore;
  36 import org.onosproject.net.intent.IntentStoreDelegate;
  37 import org.onosproject.net.intent.Key;
  38 import org.onosproject.net.intent.PartitionService;
  39 import org.onosproject.store.AbstractStore;
  40 import org.onosproject.store.service.MultiValuedTimestamp;
  41 import org.onosproject.store.service.WallClockTimestamp;
  42 import org.onosproject.store.serializers.KryoNamespaces;
  43 import org.onosproject.store.service.EventuallyConsistentMap;
  44 import org.onosproject.store.service.EventuallyConsistentMapEvent;
  45 import org.onosproject.store.service.EventuallyConsistentMapListener;
  46 import org.onosproject.store.service.StorageService;
  47 import org.slf4j.Logger;
  48
  49 import java.util.Collection;
  50 import java.util.List;
  51 import java.util.Objects;
  52 import java.util.concurrent.atomic.AtomicLong;
  53 import java.util.stream.Collectors;
  54
  55 import static com.google.common.base.Preconditions.checkNotNull;
  56 import static org.onosproject.net.intent.IntentState.PURGE_REQ;
  57 import static org.slf4j.LoggerFactory.getLogger;
  58
  59 /**
  60  * Manages inventory of Intents in a distributed data store that uses optimistic
  61  * replication and gossip based techniques.
  62  */
  63 //FIXME we should listen for leadership changes. if the local instance has just
  64 // ...  become a leader, scan the pending map and process those
  65 @Component(immediate = true, enabled = true)
  66 @Service
  67 public class GossipIntentStore
  68         extends AbstractStore<IntentEvent, IntentStoreDelegate>
  69         implements IntentStore {
  70
  71     private final Logger log = getLogger(getClass());
  72
  73     // Map of intent key => current intent state
  74     private EventuallyConsistentMap<Key, IntentData> currentMap;
  75
  76     // Map of intent key => pending intent operation
  77     private EventuallyConsistentMap<Key, IntentData> pendingMap;
  78
  79     @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
  80     protected ClusterService clusterService;
  81
  82     @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
  83     protected StorageService storageService;
  84
  85     @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
  86     protected PartitionService partitionService;
  87
  88     private final AtomicLong sequenceNumber = new AtomicLong(0);
  89
  90     @Activate
  91     public void activate() {
  92         KryoNamespace.Builder intentSerializer = KryoNamespace.newBuilder()
  93                 .register(KryoNamespaces.API)
  94                 .register(IntentData.class)
  95                 .register(MultiValuedTimestamp.class)
  96                 .register(WallClockTimestamp.class);
  97
  98         currentMap = storageService.<Key, IntentData>eventuallyConsistentMapBuilder()
  99                 .withName("intent-current")
 100                 .withSerializer(intentSerializer)
 101                 .withTimestampProvider((key, intentData) ->
 102                                             new MultiValuedTimestamp<>(intentData.version(),
 103                                                                        sequenceNumber.getAndIncrement()))
 104                 .withPeerUpdateFunction((key, intentData) -> getPeerNodes(key, intentData))
 105                 .build();
 106
 107         pendingMap = storageService.<Key, IntentData>eventuallyConsistentMapBuilder()
 108                 .withName("intent-pending")
 109                 .withSerializer(intentSerializer)
 110                 .withTimestampProvider((key, intentData) -> new MultiValuedTimestamp<>(intentData.version(),
 111                                                                                        System.nanoTime()))
 112                 .withPeerUpdateFunction((key, intentData) -> getPeerNodes(key, intentData))
 113                 .build();
 114
 115         currentMap.addListener(new InternalCurrentListener());
 116         pendingMap.addListener(new InternalPendingListener());
 117
 118         log.info("Started");
 119     }
 120
 121     @Deactivate
 122     public void deactivate() {
 123         currentMap.destroy();
 124         pendingMap.destroy();
 125
 126         log.info("Stopped");
 127     }
 128
 129     @Override
 130     public long getIntentCount() {
 131         return currentMap.size();
 132     }
 133
 134     @Override
 135     public Iterable<Intent> getIntents() {
 136         return currentMap.values().stream()
 137                 .map(IntentData::intent)
 138                 .collect(Collectors.toList());
 139     }
 140
 141     @Override
 142     public Iterable<IntentData> getIntentData(boolean localOnly, long olderThan) {
 143         if (localOnly || olderThan > 0) {
 144             long now = System.currentTimeMillis();
 145             final WallClockTimestamp time = new WallClockTimestamp(now - olderThan);
 146             return currentMap.values().stream()
 147                     .filter(data -> data.version().isOlderThan(time) &&
 148                             (!localOnly || isMaster(data.key())))
 149                     .collect(Collectors.toList());
 150         }
 151         return currentMap.values();
 152     }
 153
 154     @Override
 155     public IntentState getIntentState(Key intentKey) {
 156         IntentData data = currentMap.get(intentKey);
 157         if (data != null) {
 158             return data.state();
 159         }
 160         return null;
 161     }
 162
 163     @Override
 164     public List<Intent> getInstallableIntents(Key intentKey) {
 165         IntentData data = currentMap.get(intentKey);
 166         if (data != null) {
 167             return data.installables();
 168         }
 169         return null;
 170     }
 171
 172
 173
 174     @Override
 175     public void write(IntentData newData) {
 176         checkNotNull(newData);
 177
 178         IntentData currentData = currentMap.get(newData.key());
 179         if (IntentData.isUpdateAcceptable(currentData, newData)) {
 180             // Only the master is modifying the current state. Therefore assume
 181             // this always succeeds
 182             if (newData.state() == PURGE_REQ) {
 183                 currentMap.remove(newData.key(), currentData);
 184             } else {
 185                 currentMap.put(newData.key(), new IntentData(newData));
 186             }
 187
 188             // if current.put succeeded
 189             pendingMap.remove(newData.key(), newData);
 190         }
 191     }
 192
 193     private Collection<NodeId> getPeerNodes(Key key, IntentData data) {
 194         NodeId master = partitionService.getLeader(key);
 195         NodeId origin = (data != null) ? data.origin() : null;
 196         if (master == null || origin == null) {
 197             log.debug("Intent {} missing master and/or origin; master = {}, origin = {}",
 198                       key, master, origin);
 199         }
 200
 201         NodeId me = clusterService.getLocalNode().id();
 202         boolean isMaster = Objects.equals(master, me);
 203         boolean isOrigin = Objects.equals(origin, me);
 204         if (isMaster && isOrigin) {
 205             return getRandomNode();
 206         } else if (isMaster) {
 207             return origin != null ? ImmutableList.of(origin) : getRandomNode();
 208         } else if (isOrigin) {
 209             return master != null ? ImmutableList.of(master) : getRandomNode();
 210         } else {
 211             log.warn("No master or origin for intent {}", key);
 212             return master != null ? ImmutableList.of(master) : getRandomNode();
 213         }
 214     }
 215
 216     private List<NodeId> getRandomNode() {
 217         NodeId me = clusterService.getLocalNode().id();
 218         List<NodeId> nodes = clusterService.getNodes().stream()
 219                 .map(ControllerNode::id)
 220                 .filter(node -> !Objects.equals(node, me))
 221                 .collect(Collectors.toList());
 222         if (nodes.size() == 0) {
 223             return null;
 224         }
 225         return ImmutableList.of(nodes.get(RandomUtils.nextInt(nodes.size())));
 226     }
 227
 228     @Override
 229     public void batchWrite(Iterable<IntentData> updates) {
 230         updates.forEach(this::write);
 231     }
 232
 233     @Override
 234     public Intent getIntent(Key key) {
 235         IntentData data = currentMap.get(key);
 236         if (data != null) {
 237             return data.intent();
 238         }
 239         return null;
 240     }
 241
 242     @Override
 243     public IntentData getIntentData(Key key) {
 244         IntentData current = currentMap.get(key);
 245         if (current == null) {
 246             return null;
 247         }
 248         return new IntentData(current);
 249     }
 250
 251     @Override
 252     public void addPending(IntentData data) {
 253         checkNotNull(data);
 254
 255         if (data.version() == null) {
 256             data.setVersion(new WallClockTimestamp());
 257         }
 258         data.setOrigin(clusterService.getLocalNode().id());
 259         pendingMap.put(data.key(), new IntentData(data));
 260     }
 261
 262     @Override
 263     public boolean isMaster(Key intentKey) {
 264         return partitionService.isMine(intentKey);
 265     }
 266
 267     @Override
 268     public Iterable<Intent> getPending() {
 269         return pendingMap.values().stream()
 270                 .map(IntentData::intent)
 271                 .collect(Collectors.toList());
 272     }
 273
 274     @Override
 275     public Iterable<IntentData> getPendingData() {
 276         return pendingMap.values();
 277     }
 278
 279     @Override
 280     public Iterable<IntentData> getPendingData(boolean localOnly, long olderThan) {
 281         long now = System.currentTimeMillis();
 282         final WallClockTimestamp time = new WallClockTimestamp(now - olderThan);
 283         return pendingMap.values().stream()
 284                 .filter(data -> data.version().isOlderThan(time) &&
 285                                 (!localOnly || isMaster(data.key())))
 286                 .collect(Collectors.toList());
 287     }
 288
 289     private void notifyDelegateIfNotNull(IntentEvent event) {
 290         if (event != null) {
 291             notifyDelegate(event);
 292         }
 293     }
 294
 295     private final class InternalCurrentListener implements
 296             EventuallyConsistentMapListener<Key, IntentData> {
 297         @Override
 298         public void event(EventuallyConsistentMapEvent<Key, IntentData> event) {
 299             IntentData intentData = event.value();
 300
 301             if (event.type() == EventuallyConsistentMapEvent.Type.PUT) {
 302                 // The current intents map has been updated. If we are master for
 303                 // this intent's partition, notify the Manager that it should
 304                 // emit notifications about updated tracked resources.
 305                 if (delegate != null && isMaster(event.value().intent().key())) {
 306                     delegate.onUpdate(new IntentData(intentData)); // copy for safety, likely unnecessary
 307                 }
 308                 notifyDelegateIfNotNull(IntentEvent.getEvent(intentData));
 309             }
 310         }
 311     }
 312
 313     private final class InternalPendingListener implements
 314             EventuallyConsistentMapListener<Key, IntentData> {
 315         @Override
 316         public void event(
 317                 EventuallyConsistentMapEvent<Key, IntentData> event) {
 318             if (event.type() == EventuallyConsistentMapEvent.Type.PUT) {
 319                 // The pending intents map has been updated. If we are master for
 320                 // this intent's partition, notify the Manager that it should do
 321                 // some work.
 322                 if (isMaster(event.value().intent().key())) {
 323                     if (delegate != null) {
 324                         delegate.process(new IntentData(event.value()));
 325                     }
 326                 }
 327
 328                 notifyDelegateIfNotNull(IntentEvent.getEvent(event.value()));
 329             }
 330         }
 331     }
 332
 333 }
 334