opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/utils/ActorContext.java

   1 /*
   2  * Copyright (c) 2014 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.datastore.utils;
  10
  11 import static akka.pattern.Patterns.ask;
  12 import akka.actor.ActorPath;
  13 import akka.actor.ActorRef;
  14 import akka.actor.ActorSelection;
  15 import akka.actor.ActorSystem;
  16 import akka.actor.Address;
  17 import akka.actor.PoisonPill;
  18 import akka.dispatch.Futures;
  19 import akka.dispatch.Mapper;
  20 import akka.dispatch.OnComplete;
  21 import akka.pattern.AskTimeoutException;
  22 import akka.util.Timeout;
  23 import com.codahale.metrics.MetricRegistry;
  24 import com.codahale.metrics.Timer;
  25 import com.google.common.annotations.VisibleForTesting;
  26 import com.google.common.base.Optional;
  27 import com.google.common.base.Preconditions;
  28 import com.google.common.base.Strings;
  29 import com.google.common.cache.Cache;
  30 import com.google.common.cache.CacheBuilder;
  31 import com.google.common.util.concurrent.RateLimiter;
  32 import java.util.concurrent.TimeUnit;
  33 import org.opendaylight.controller.cluster.common.actor.CommonConfig;
  34 import org.opendaylight.controller.cluster.datastore.ClusterWrapper;
  35 import org.opendaylight.controller.cluster.datastore.Configuration;
  36 import org.opendaylight.controller.cluster.datastore.DatastoreContext;
  37 import org.opendaylight.controller.cluster.datastore.exceptions.LocalShardNotFoundException;
  38 import org.opendaylight.controller.cluster.datastore.exceptions.NoShardLeaderException;
  39 import org.opendaylight.controller.cluster.datastore.exceptions.NotInitializedException;
  40 import org.opendaylight.controller.cluster.datastore.exceptions.PrimaryNotFoundException;
  41 import org.opendaylight.controller.cluster.datastore.exceptions.TimeoutException;
  42 import org.opendaylight.controller.cluster.datastore.exceptions.UnknownMessageException;
  43 import org.opendaylight.controller.cluster.datastore.messages.FindLocalShard;
  44 import org.opendaylight.controller.cluster.datastore.messages.FindPrimary;
  45 import org.opendaylight.controller.cluster.datastore.messages.LocalPrimaryShardFound;
  46 import org.opendaylight.controller.cluster.datastore.messages.LocalShardFound;
  47 import org.opendaylight.controller.cluster.datastore.messages.LocalShardNotFound;
  48 import org.opendaylight.controller.cluster.datastore.messages.PrimaryShardInfo;
  49 import org.opendaylight.controller.cluster.datastore.messages.RemotePrimaryShardFound;
  50 import org.opendaylight.controller.cluster.datastore.messages.UpdateSchemaContext;
  51 import org.opendaylight.controller.cluster.reporting.MetricsReporter;
  52 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTree;
  53 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
  54 import org.slf4j.Logger;
  55 import org.slf4j.LoggerFactory;
  56 import scala.concurrent.Await;
  57 import scala.concurrent.ExecutionContext;
  58 import scala.concurrent.Future;
  59 import scala.concurrent.duration.Duration;
  60 import scala.concurrent.duration.FiniteDuration;
  61
  62 /**
  63  * The ActorContext class contains utility methods which could be used by
  64  * non-actors (like DistributedDataStore) to work with actors a little more
  65  * easily. An ActorContext can be freely passed around to local object instances
  66  * but should not be passed to actors especially remote actors
  67  */
  68 public class ActorContext {
  69     private static final Logger LOG = LoggerFactory.getLogger(ActorContext.class);
  70     private static final String DISTRIBUTED_DATA_STORE_METRIC_REGISTRY = "distributed-data-store";
  71     private static final String METRIC_RATE = "rate";
  72     private static final Mapper<Throwable, Throwable> FIND_PRIMARY_FAILURE_TRANSFORMER =
  73                                                               new Mapper<Throwable, Throwable>() {
  74         @Override
  75         public Throwable apply(Throwable failure) {
  76             Throwable actualFailure = failure;
  77             if(failure instanceof AskTimeoutException) {
  78                 // A timeout exception most likely means the shard isn't initialized.
  79                 actualFailure = new NotInitializedException(
  80                         "Timed out trying to find the primary shard. Most likely cause is the " +
  81                         "shard is not initialized yet.");
  82             }
  83
  84             return actualFailure;
  85         }
  86     };
  87     public static final String MAILBOX = "bounded-mailbox";
  88
  89     private final ActorSystem actorSystem;
  90     private final ActorRef shardManager;
  91     private final ClusterWrapper clusterWrapper;
  92     private final Configuration configuration;
  93     private DatastoreContext datastoreContext;
  94     private FiniteDuration operationDuration;
  95     private Timeout operationTimeout;
  96     private final String selfAddressHostPort;
  97     private RateLimiter txRateLimiter;
  98     private final int transactionOutstandingOperationLimit;
  99     private Timeout transactionCommitOperationTimeout;
 100     private Timeout shardInitializationTimeout;
 101     private final Dispatchers dispatchers;
 102     private Cache<String, Future<PrimaryShardInfo>> primaryShardInfoCache;
 103
 104     private volatile SchemaContext schemaContext;
 105     private volatile boolean updated;
 106     private final MetricRegistry metricRegistry = MetricsReporter.getInstance(DatastoreContext.METRICS_DOMAIN).getMetricsRegistry();
 107
 108     public ActorContext(ActorSystem actorSystem, ActorRef shardManager,
 109             ClusterWrapper clusterWrapper, Configuration configuration) {
 110         this(actorSystem, shardManager, clusterWrapper, configuration,
 111                 DatastoreContext.newBuilder().build());
 112     }
 113
 114     public ActorContext(ActorSystem actorSystem, ActorRef shardManager,
 115             ClusterWrapper clusterWrapper, Configuration configuration,
 116             DatastoreContext datastoreContext) {
 117         this.actorSystem = actorSystem;
 118         this.shardManager = shardManager;
 119         this.clusterWrapper = clusterWrapper;
 120         this.configuration = configuration;
 121         this.datastoreContext = datastoreContext;
 122         this.dispatchers = new Dispatchers(actorSystem.dispatchers());
 123
 124         setCachedProperties();
 125
 126         Address selfAddress = clusterWrapper.getSelfAddress();
 127         if (selfAddress != null && !selfAddress.host().isEmpty()) {
 128             selfAddressHostPort = selfAddress.host().get() + ":" + selfAddress.port().get();
 129         } else {
 130             selfAddressHostPort = null;
 131         }
 132
 133         transactionOutstandingOperationLimit = new CommonConfig(this.getActorSystem().settings().config()).getMailBoxCapacity();
 134     }
 135
 136     private void setCachedProperties() {
 137         txRateLimiter = RateLimiter.create(datastoreContext.getTransactionCreationInitialRateLimit());
 138
 139         operationDuration = Duration.create(datastoreContext.getOperationTimeoutInSeconds(), TimeUnit.SECONDS);
 140         operationTimeout = new Timeout(operationDuration);
 141
 142         transactionCommitOperationTimeout =  new Timeout(Duration.create(
 143                 datastoreContext.getShardTransactionCommitTimeoutInSeconds(), TimeUnit.SECONDS));
 144
 145         shardInitializationTimeout = new Timeout(datastoreContext.getShardInitializationTimeout().duration().$times(2));
 146
 147         primaryShardInfoCache = CacheBuilder.newBuilder()
 148                 .expireAfterWrite(datastoreContext.getShardLeaderElectionTimeout().duration().toMillis(), TimeUnit.MILLISECONDS)
 149                 .build();
 150     }
 151
 152     public DatastoreContext getDatastoreContext() {
 153         return datastoreContext;
 154     }
 155
 156     public ActorSystem getActorSystem() {
 157         return actorSystem;
 158     }
 159
 160     public ActorRef getShardManager() {
 161         return shardManager;
 162     }
 163
 164     public ActorSelection actorSelection(String actorPath) {
 165         return actorSystem.actorSelection(actorPath);
 166     }
 167
 168     public ActorSelection actorSelection(ActorPath actorPath) {
 169         return actorSystem.actorSelection(actorPath);
 170     }
 171
 172     public void setSchemaContext(SchemaContext schemaContext) {
 173         this.schemaContext = schemaContext;
 174
 175         if(shardManager != null) {
 176             shardManager.tell(new UpdateSchemaContext(schemaContext), ActorRef.noSender());
 177         }
 178     }
 179
 180     public void setDatastoreContext(DatastoreContext context) {
 181         this.datastoreContext = context;
 182         setCachedProperties();
 183
 184         // We write the 'updated' volatile to trigger a write memory barrier so that the writes above
 185         // will be published immediately even though they may not be immediately visible to other
 186         // threads due to unsynchronized reads. That's OK though - we're going for eventual
 187         // consistency here as immediately visible updates to these members aren't critical. These
 188         // members could've been made volatile but wanted to avoid volatile reads as these are
 189         // accessed often and updates will be infrequent.
 190
 191         updated = true;
 192
 193         if(shardManager != null) {
 194             shardManager.tell(context, ActorRef.noSender());
 195         }
 196     }
 197
 198     public SchemaContext getSchemaContext() {
 199         return schemaContext;
 200     }
 201
 202     public Future<PrimaryShardInfo> findPrimaryShardAsync(final String shardName) {
 203         Future<PrimaryShardInfo> ret = primaryShardInfoCache.getIfPresent(shardName);
 204         if(ret != null){
 205             return ret;
 206         }
 207         Future<Object> future = executeOperationAsync(shardManager,
 208                 new FindPrimary(shardName, true), shardInitializationTimeout);
 209
 210         return future.transform(new Mapper<Object, PrimaryShardInfo>() {
 211             @Override
 212             public PrimaryShardInfo checkedApply(Object response) throws Exception {
 213                 if(response instanceof RemotePrimaryShardFound) {
 214                     LOG.debug("findPrimaryShardAsync received: {}", response);
 215                     return onPrimaryShardFound(shardName, ((RemotePrimaryShardFound)response).getPrimaryPath(), null);
 216                 } else if(response instanceof LocalPrimaryShardFound) {
 217                     LOG.debug("findPrimaryShardAsync received: {}", response);
 218                     LocalPrimaryShardFound found = (LocalPrimaryShardFound)response;
 219                     return onPrimaryShardFound(shardName, found.getPrimaryPath(), found.getLocalShardDataTree());
 220                 } else if(response instanceof NotInitializedException) {
 221                     throw (NotInitializedException)response;
 222                 } else if(response instanceof PrimaryNotFoundException) {
 223                     throw (PrimaryNotFoundException)response;
 224                 } else if(response instanceof NoShardLeaderException) {
 225                     throw (NoShardLeaderException)response;
 226                 }
 227
 228                 throw new UnknownMessageException(String.format(
 229                         "FindPrimary returned unkown response: %s", response));
 230             }
 231         }, FIND_PRIMARY_FAILURE_TRANSFORMER, getClientDispatcher());
 232     }
 233
 234     private PrimaryShardInfo onPrimaryShardFound(String shardName, String primaryActorPath,
 235             DataTree localShardDataTree) {
 236         ActorSelection actorSelection = actorSystem.actorSelection(primaryActorPath);
 237         PrimaryShardInfo info = new PrimaryShardInfo(actorSelection, Optional.fromNullable(localShardDataTree));
 238         primaryShardInfoCache.put(shardName, Futures.successful(info));
 239         return info;
 240     }
 241
 242     /**
 243      * Finds a local shard given its shard name and return it's ActorRef
 244      *
 245      * @param shardName the name of the local shard that needs to be found
 246      * @return a reference to a local shard actor which represents the shard
 247      *         specified by the shardName
 248      */
 249     public Optional<ActorRef> findLocalShard(String shardName) {
 250         Object result = executeOperation(shardManager, new FindLocalShard(shardName, false));
 251
 252         if (result instanceof LocalShardFound) {
 253             LocalShardFound found = (LocalShardFound) result;
 254             LOG.debug("Local shard found {}", found.getPath());
 255             return Optional.of(found.getPath());
 256         }
 257
 258         return Optional.absent();
 259     }
 260
 261     /**
 262      * Finds a local shard async given its shard name and return a Future from which to obtain the
 263      * ActorRef.
 264      *
 265      * @param shardName the name of the local shard that needs to be found
 266      */
 267     public Future<ActorRef> findLocalShardAsync( final String shardName) {
 268         Future<Object> future = executeOperationAsync(shardManager,
 269                 new FindLocalShard(shardName, true), shardInitializationTimeout);
 270
 271         return future.map(new Mapper<Object, ActorRef>() {
 272             @Override
 273             public ActorRef checkedApply(Object response) throws Throwable {
 274                 if(response instanceof LocalShardFound) {
 275                     LocalShardFound found = (LocalShardFound)response;
 276                     LOG.debug("Local shard found {}", found.getPath());
 277                     return found.getPath();
 278                 } else if(response instanceof NotInitializedException) {
 279                     throw (NotInitializedException)response;
 280                 } else if(response instanceof LocalShardNotFound) {
 281                     throw new LocalShardNotFoundException(
 282                             String.format("Local shard for %s does not exist.", shardName));
 283                 }
 284
 285                 throw new UnknownMessageException(String.format(
 286                         "FindLocalShard returned unkown response: %s", response));
 287             }
 288         }, getClientDispatcher());
 289     }
 290
 291     /**
 292      * Executes an operation on a local actor and wait for it's response
 293      *
 294      * @param actor
 295      * @param message
 296      * @return The response of the operation
 297      */
 298     public Object executeOperation(ActorRef actor, Object message) {
 299         Future<Object> future = executeOperationAsync(actor, message, operationTimeout);
 300
 301         try {
 302             return Await.result(future, operationDuration);
 303         } catch (Exception e) {
 304             throw new TimeoutException("Sending message " + message.getClass().toString() +
 305                     " to actor " + actor.toString() + " failed. Try again later.", e);
 306         }
 307     }
 308
 309     public Future<Object> executeOperationAsync(ActorRef actor, Object message, Timeout timeout) {
 310         Preconditions.checkArgument(actor != null, "actor must not be null");
 311         Preconditions.checkArgument(message != null, "message must not be null");
 312
 313         LOG.debug("Sending message {} to {}", message.getClass(), actor);
 314         return doAsk(actor, message, timeout);
 315     }
 316
 317     /**
 318      * Execute an operation on a remote actor and wait for it's response
 319      *
 320      * @param actor
 321      * @param message
 322      * @return
 323      */
 324     public Object executeOperation(ActorSelection actor, Object message) {
 325         Future<Object> future = executeOperationAsync(actor, message);
 326
 327         try {
 328             return Await.result(future, operationDuration);
 329         } catch (Exception e) {
 330             throw new TimeoutException("Sending message " + message.getClass().toString() +
 331                     " to actor " + actor.toString() + " failed. Try again later.", e);
 332         }
 333     }
 334
 335     /**
 336      * Execute an operation on a remote actor asynchronously.
 337      *
 338      * @param actor the ActorSelection
 339      * @param message the message to send
 340      * @param timeout the operation timeout
 341      * @return a Future containing the eventual result
 342      */
 343     public Future<Object> executeOperationAsync(ActorSelection actor, Object message,
 344             Timeout timeout) {
 345         Preconditions.checkArgument(actor != null, "actor must not be null");
 346         Preconditions.checkArgument(message != null, "message must not be null");
 347
 348         LOG.debug("Sending message {} to {}", message.getClass(), actor);
 349
 350         return doAsk(actor, message, timeout);
 351     }
 352
 353     /**
 354      * Execute an operation on a remote actor asynchronously.
 355      *
 356      * @param actor the ActorSelection
 357      * @param message the message to send
 358      * @return a Future containing the eventual result
 359      */
 360     public Future<Object> executeOperationAsync(ActorSelection actor, Object message) {
 361         return executeOperationAsync(actor, message, operationTimeout);
 362     }
 363
 364     /**
 365      * Sends an operation to be executed by a remote actor asynchronously without waiting for a
 366      * reply (essentially set and forget).
 367      *
 368      * @param actor the ActorSelection
 369      * @param message the message to send
 370      */
 371     public void sendOperationAsync(ActorSelection actor, Object message) {
 372         Preconditions.checkArgument(actor != null, "actor must not be null");
 373         Preconditions.checkArgument(message != null, "message must not be null");
 374
 375         LOG.debug("Sending message {} to {}", message.getClass(), actor);
 376
 377         actor.tell(message, ActorRef.noSender());
 378     }
 379
 380     public void shutdown() {
 381         shardManager.tell(PoisonPill.getInstance(), ActorRef.noSender());
 382     }
 383
 384     public ClusterWrapper getClusterWrapper() {
 385         return clusterWrapper;
 386     }
 387
 388     public String getCurrentMemberName(){
 389         return clusterWrapper.getCurrentMemberName();
 390     }
 391
 392     /**
 393      * Send the message to each and every shard
 394      *
 395      * @param message
 396      */
 397     public void broadcast(final Object message){
 398         for(final String shardName : configuration.getAllShardNames()){
 399
 400             Future<PrimaryShardInfo> primaryFuture = findPrimaryShardAsync(shardName);
 401             primaryFuture.onComplete(new OnComplete<PrimaryShardInfo>() {
 402                 @Override
 403                 public void onComplete(Throwable failure, PrimaryShardInfo primaryShardInfo) {
 404                     if(failure != null) {
 405                         LOG.warn("broadcast failed to send message {} to shard {}:  {}",
 406                                 message.getClass().getSimpleName(), shardName, failure);
 407                     } else {
 408                         primaryShardInfo.getPrimaryShardActor().tell(message, ActorRef.noSender());
 409                     }
 410                 }
 411             }, getClientDispatcher());
 412         }
 413     }
 414
 415     public FiniteDuration getOperationDuration() {
 416         return operationDuration;
 417     }
 418
 419     public boolean isPathLocal(String path) {
 420         if (Strings.isNullOrEmpty(path)) {
 421             return false;
 422         }
 423
 424         int pathAtIndex = path.indexOf('@');
 425         if (pathAtIndex == -1) {
 426             //if the path is of local format, then its local and is co-located
 427             return true;
 428
 429         } else if (selfAddressHostPort != null) {
 430             // self-address and tx actor path, both are of remote path format
 431             int slashIndex = path.indexOf('/', pathAtIndex);
 432
 433             if (slashIndex == -1) {
 434                 return false;
 435             }
 436
 437             String hostPort = path.substring(pathAtIndex + 1, slashIndex);
 438             return hostPort.equals(selfAddressHostPort);
 439
 440         } else {
 441             // self address is local format and tx actor path is remote format
 442             return false;
 443         }
 444     }
 445
 446     /**
 447      * @deprecated This method is present only to support backward compatibility with Helium and should not be
 448      * used any further
 449      *
 450      *
 451      * @param primaryPath
 452      * @param localPathOfRemoteActor
 453      * @return
 454     */
 455     @Deprecated
 456     public String resolvePath(final String primaryPath,
 457                                             final String localPathOfRemoteActor) {
 458         StringBuilder builder = new StringBuilder();
 459         String[] primaryPathElements = primaryPath.split("/");
 460         builder.append(primaryPathElements[0]).append("//")
 461             .append(primaryPathElements[1]).append(primaryPathElements[2]);
 462         String[] remotePathElements = localPathOfRemoteActor.split("/");
 463         for (int i = 3; i < remotePathElements.length; i++) {
 464                 builder.append("/").append(remotePathElements[i]);
 465             }
 466
 467         return builder.toString();
 468     }
 469
 470     /**
 471      * Get the maximum number of operations that are to be permitted within a transaction before the transaction
 472      * should begin throttling the operations
 473      *
 474      * Parking reading this configuration here because we need to get to the actor system settings
 475      *
 476      * @return
 477      */
 478     public int getTransactionOutstandingOperationLimit(){
 479         return transactionOutstandingOperationLimit;
 480     }
 481
 482     /**
 483      * This is a utility method that lets us get a Timer object for any operation. This is a little open-ended to allow
 484      * us to create a timer for pretty much anything.
 485      *
 486      * @param operationName
 487      * @return
 488      */
 489     public Timer getOperationTimer(String operationName){
 490         return getOperationTimer(datastoreContext.getDataStoreType(), operationName);
 491     }
 492
 493     public Timer getOperationTimer(String dataStoreType, String operationName){
 494         final String rate = MetricRegistry.name(DISTRIBUTED_DATA_STORE_METRIC_REGISTRY, dataStoreType,
 495                 operationName, METRIC_RATE);
 496         return metricRegistry.timer(rate);
 497     }
 498
 499     /**
 500      * Get the type of the data store to which this ActorContext belongs
 501      *
 502      * @return
 503      */
 504     public String getDataStoreType() {
 505         return datastoreContext.getDataStoreType();
 506     }
 507
 508     /**
 509      * Set the number of transaction creation permits that are to be allowed
 510      *
 511      * @param permitsPerSecond
 512      */
 513     public void setTxCreationLimit(double permitsPerSecond){
 514         txRateLimiter.setRate(permitsPerSecond);
 515     }
 516
 517     /**
 518      * Get the current transaction creation rate limit
 519      * @return
 520      */
 521     public double getTxCreationLimit(){
 522         return txRateLimiter.getRate();
 523     }
 524
 525     /**
 526      * Try to acquire a transaction creation permit. Will block if no permits are available.
 527      */
 528     public void acquireTxCreationPermit(){
 529         txRateLimiter.acquire();
 530     }
 531
 532     /**
 533      * Return the operation timeout to be used when committing transactions
 534      * @return
 535      */
 536     public Timeout getTransactionCommitOperationTimeout(){
 537         return transactionCommitOperationTimeout;
 538     }
 539
 540     /**
 541      * An akka dispatcher that is meant to be used when processing ask Futures which were triggered by client
 542      * code on the datastore
 543      * @return
 544      */
 545     public ExecutionContext getClientDispatcher() {
 546         return this.dispatchers.getDispatcher(Dispatchers.DispatcherType.Client);
 547     }
 548
 549     public String getNotificationDispatcherPath(){
 550         return this.dispatchers.getDispatcherPath(Dispatchers.DispatcherType.Notification);
 551     }
 552
 553     public Configuration getConfiguration() {
 554         return configuration;
 555     }
 556
 557     protected Future<Object> doAsk(ActorRef actorRef, Object message, Timeout timeout){
 558         return ask(actorRef, message, timeout);
 559     }
 560
 561     protected Future<Object> doAsk(ActorSelection actorRef, Object message, Timeout timeout){
 562         return ask(actorRef, message, timeout);
 563     }
 564
 565     @VisibleForTesting
 566     Cache<String, Future<PrimaryShardInfo>> getPrimaryShardInfoCache() {
 567         return primaryShardInfoCache;
 568     }
 569 }