opendaylight/md-sal/eos-dom-akka/src/main/java/org/opendaylight/controller/eos/akka/owner/supervisor/OwnerSyncer.java

   1 /*
   2  * Copyright (c) 2021 PANTHEON.tech, s.r.o. 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.eos.akka.owner.supervisor;
   9
  10 import akka.actor.typed.ActorRef;
  11 import akka.actor.typed.Behavior;
  12 import akka.actor.typed.javadsl.AbstractBehavior;
  13 import akka.actor.typed.javadsl.ActorContext;
  14 import akka.actor.typed.javadsl.Behaviors;
  15 import akka.actor.typed.javadsl.Receive;
  16 import akka.cluster.ddata.LWWRegister;
  17 import akka.cluster.ddata.LWWRegisterKey;
  18 import akka.cluster.ddata.ORMap;
  19 import akka.cluster.ddata.ORSet;
  20 import akka.cluster.ddata.typed.javadsl.DistributedData;
  21 import akka.cluster.ddata.typed.javadsl.Replicator;
  22 import akka.cluster.ddata.typed.javadsl.ReplicatorMessageAdapter;
  23 import java.time.Duration;
  24 import java.util.HashMap;
  25 import java.util.HashSet;
  26 import java.util.Map;
  27 import java.util.Set;
  28 import org.eclipse.jdt.annotation.Nullable;
  29 import org.opendaylight.controller.eos.akka.owner.supervisor.command.DataCenterActivated;
  30 import org.opendaylight.controller.eos.akka.owner.supervisor.command.InitialCandidateSync;
  31 import org.opendaylight.controller.eos.akka.owner.supervisor.command.InitialOwnerSync;
  32 import org.opendaylight.controller.eos.akka.owner.supervisor.command.OwnerSupervisorCommand;
  33 import org.opendaylight.controller.eos.akka.owner.supervisor.command.OwnerSupervisorReply;
  34 import org.opendaylight.controller.eos.akka.registry.candidate.CandidateRegistry;
  35 import org.opendaylight.mdsal.eos.dom.api.DOMEntity;
  36 import org.slf4j.Logger;
  37 import org.slf4j.LoggerFactory;
  38
  39 /**
  40  * Behavior that retrieves current candidates/owners from distributed-data and switches to OwnerSupervisor when the
  41  * sync has finished.
  42  */
  43 public final class OwnerSyncer extends AbstractBehavior<OwnerSupervisorCommand> {
  44     private static final Logger LOG = LoggerFactory.getLogger(OwnerSyncer.class);
  45
  46     private final ReplicatorMessageAdapter<OwnerSupervisorCommand, LWWRegister<String>> ownerReplicator;
  47     private final Map<DOMEntity, Set<String>> currentCandidates = new HashMap<>();
  48     private final Map<DOMEntity, String> currentOwners = new HashMap<>();
  49
  50     // String representation of Entity to DOMEntity
  51     private final Map<String, DOMEntity> entityLookup = new HashMap<>();
  52
  53     private int toSync = -1;
  54
  55     private OwnerSyncer(final ActorContext<OwnerSupervisorCommand> context,
  56                         @Nullable final ActorRef<OwnerSupervisorReply> notifyDatacenterStarted) {
  57         super(context);
  58         LOG.debug("Starting candidate and owner sync");
  59
  60         final ActorRef<Replicator.Command> replicator = DistributedData.get(context.getSystem()).replicator();
  61
  62         this.ownerReplicator = new ReplicatorMessageAdapter<>(context, replicator, Duration.ofSeconds(5));
  63
  64         new ReplicatorMessageAdapter<OwnerSupervisorCommand, ORMap<DOMEntity, ORSet<String>>>(context, replicator,
  65             Duration.ofSeconds(5)).askGet(
  66                 askReplyTo -> new Replicator.Get<>(CandidateRegistry.KEY, Replicator.readLocal(), askReplyTo),
  67                 InitialCandidateSync::new);
  68
  69         if (notifyDatacenterStarted != null) {
  70             notifyDatacenterStarted.tell(DataCenterActivated.INSTANCE);
  71         }
  72     }
  73
  74     public static Behavior<OwnerSupervisorCommand> create(
  75             final ActorRef<OwnerSupervisorReply> notifyDatacenterStarted) {
  76         return Behaviors.setup(ctx -> new OwnerSyncer(ctx, notifyDatacenterStarted));
  77     }
  78
  79     @Override
  80     public Receive<OwnerSupervisorCommand> createReceive() {
  81         return newReceiveBuilder()
  82                 .onMessage(InitialCandidateSync.class, this::onInitialCandidateSync)
  83                 .onMessage(InitialOwnerSync.class, this::onInitialOwnerSync)
  84                 .build();
  85     }
  86
  87     private Behavior<OwnerSupervisorCommand> onInitialCandidateSync(final InitialCandidateSync rsp) {
  88         final Replicator.GetResponse<ORMap<DOMEntity, ORSet<String>>> response = rsp.getResponse();
  89         if (response instanceof Replicator.GetSuccess) {
  90             return doInitialSync((Replicator.GetSuccess<ORMap<DOMEntity, ORSet<String>>>) response);
  91         } else if (response instanceof Replicator.NotFound) {
  92             LOG.debug("No candidates found switching to supervisor");
  93             return switchToSupervisor();
  94         } else {
  95             LOG.debug("Initial candidate sync failed, switching to supervisor. Sync reply: {}", response);
  96             return switchToSupervisor();
  97         }
  98     }
  99
 100     private Behavior<OwnerSupervisorCommand> doInitialSync(
 101             final Replicator.GetSuccess<ORMap<DOMEntity, ORSet<String>>> response) {
 102
 103         final ORMap<DOMEntity, ORSet<String>> candidates = response.get(CandidateRegistry.KEY);
 104         candidates.getEntries().entrySet().forEach(entry -> {
 105             currentCandidates.put(entry.getKey(), new HashSet<>(entry.getValue().getElements()));
 106         });
 107
 108         toSync = candidates.keys().size();
 109         for (final DOMEntity entity : candidates.keys().getElements()) {
 110             entityLookup.put(entity.toString(), entity);
 111
 112             ownerReplicator.askGet(
 113                     askReplyTo -> new Replicator.Get<>(
 114                             new LWWRegisterKey<>(entity.toString()),
 115                             Replicator.readLocal(),
 116                             askReplyTo),
 117                     InitialOwnerSync::new);
 118         }
 119
 120         return this;
 121     }
 122
 123     private Behavior<OwnerSupervisorCommand> onInitialOwnerSync(final InitialOwnerSync rsp) {
 124         final Replicator.GetResponse<LWWRegister<String>> response = rsp.getResponse();
 125         if (response instanceof Replicator.GetSuccess) {
 126             handleOwnerRsp((Replicator.GetSuccess<LWWRegister<String>>) response);
 127         } else if (response instanceof Replicator.NotFound) {
 128             handleNotFoundOwnerRsp((Replicator.NotFound<LWWRegister<String>>) response);
 129         } else {
 130             LOG.debug("Initial sync failed response: {}", response);
 131         }
 132
 133         // count the responses, on last switch behaviors
 134         toSync--;
 135         if (toSync == 0) {
 136             return switchToSupervisor();
 137         }
 138
 139         return this;
 140     }
 141
 142     private Behavior<OwnerSupervisorCommand> switchToSupervisor() {
 143         LOG.debug("Initial sync done, switching to supervisor. candidates: {}, owners: {}",
 144                 currentCandidates, currentOwners);
 145         return Behaviors.setup(ctx ->
 146                 OwnerSupervisor.create(currentCandidates, currentOwners));
 147     }
 148
 149     private void handleOwnerRsp(final Replicator.GetSuccess<LWWRegister<String>> rsp) {
 150         final DOMEntity entity = entityLookup.get(rsp.key().id());
 151         final String owner = rsp.get(rsp.key()).getValue();
 152
 153         currentOwners.put(entity, owner);
 154     }
 155
 156     private static void handleNotFoundOwnerRsp(final Replicator.NotFound<LWWRegister<String>> rsp) {
 157         LOG.debug("Owner not found. {}", rsp);
 158     }
 159 }