singleton-service/mdsal-singleton-dom-impl/src/main/java/org/opendaylight/mdsal/singleton/dom/impl/EOSClusterSingletonServiceProvider.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.mdsal.singleton.dom.impl;
   9
  10 import static com.google.common.base.Preconditions.checkArgument;
  11 import static com.google.common.base.Verify.verify;
  12 import static com.google.common.base.Verify.verifyNotNull;
  13 import static java.util.Objects.requireNonNull;
  14
  15 import com.google.common.annotations.VisibleForTesting;
  16 import com.google.common.base.Strings;
  17 import com.google.common.util.concurrent.Futures;
  18 import com.google.common.util.concurrent.ListenableFuture;
  19 import com.google.common.util.concurrent.MoreExecutors;
  20 import java.util.ArrayList;
  21 import java.util.List;
  22 import java.util.concurrent.ConcurrentHashMap;
  23 import java.util.concurrent.ConcurrentMap;
  24 import java.util.concurrent.ExecutionException;
  25 import javax.annotation.PreDestroy;
  26 import javax.inject.Inject;
  27 import javax.inject.Singleton;
  28 import org.eclipse.jdt.annotation.NonNull;
  29 import org.opendaylight.mdsal.eos.common.api.CandidateAlreadyRegisteredException;
  30 import org.opendaylight.mdsal.eos.common.api.EntityOwnershipStateChange;
  31 import org.opendaylight.mdsal.eos.dom.api.DOMEntity;
  32 import org.opendaylight.mdsal.eos.dom.api.DOMEntityOwnershipListener;
  33 import org.opendaylight.mdsal.eos.dom.api.DOMEntityOwnershipService;
  34 import org.opendaylight.mdsal.singleton.common.api.ClusterSingletonService;
  35 import org.opendaylight.mdsal.singleton.common.api.ClusterSingletonServiceProvider;
  36 import org.opendaylight.mdsal.singleton.common.api.ClusterSingletonServiceRegistration;
  37 import org.opendaylight.yangtools.concepts.Registration;
  38 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.NodeIdentifierWithPredicates;
  39 import org.osgi.service.component.annotations.Activate;
  40 import org.osgi.service.component.annotations.Component;
  41 import org.osgi.service.component.annotations.Deactivate;
  42 import org.osgi.service.component.annotations.Reference;
  43 import org.slf4j.Logger;
  44 import org.slf4j.LoggerFactory;
  45
  46 /**
  47  * Implementation of {@link ClusterSingletonServiceProvider} working on top a {@link DOMEntityOwnershipService}.
  48  */
  49 @Singleton
  50 @Component(service = ClusterSingletonServiceProvider.class)
  51 public final class EOSClusterSingletonServiceProvider
  52         implements ClusterSingletonServiceProvider, DOMEntityOwnershipListener, AutoCloseable {
  53     private static final Logger LOG = LoggerFactory.getLogger(EOSClusterSingletonServiceProvider.class);
  54
  55     @VisibleForTesting
  56     static final @NonNull String SERVICE_ENTITY_TYPE = "org.opendaylight.mdsal.ServiceEntityType";
  57     @VisibleForTesting
  58     static final @NonNull String CLOSE_SERVICE_ENTITY_TYPE = "org.opendaylight.mdsal.AsyncServiceCloseEntityType";
  59
  60     private final ConcurrentMap<String, ClusterSingletonServiceGroup> serviceGroupMap = new ConcurrentHashMap<>();
  61     private final DOMEntityOwnershipService entityOwnershipService;
  62
  63     /* EOS Entity Listeners Registration */
  64     private Registration serviceEntityListenerReg;
  65     private Registration asyncCloseEntityListenerReg;
  66
  67     @Inject
  68     @Activate
  69     public EOSClusterSingletonServiceProvider(@Reference final DOMEntityOwnershipService entityOwnershipService) {
  70         this.entityOwnershipService = requireNonNull(entityOwnershipService);
  71         serviceEntityListenerReg = entityOwnershipService.registerListener(SERVICE_ENTITY_TYPE, this);
  72         asyncCloseEntityListenerReg = entityOwnershipService.registerListener(CLOSE_SERVICE_ENTITY_TYPE, this);
  73         LOG.info("Cluster Singleton Service started");
  74     }
  75
  76     @PreDestroy
  77     @Deactivate
  78     @Override
  79     public synchronized void close() throws ExecutionException, InterruptedException {
  80         if (serviceEntityListenerReg == null) {
  81             // Idempotent
  82             return;
  83         }
  84
  85         LOG.info("Cluster Singleton Service stopping");
  86         serviceEntityListenerReg.close();
  87         serviceEntityListenerReg = null;
  88
  89         final var future = Futures.allAsList(serviceGroupMap.values().stream()
  90             .map(ClusterSingletonServiceGroup::closeClusterSingletonGroup)
  91             .toList());
  92         try {
  93             LOG.debug("Waiting for service groups to stop");
  94             future.get();
  95         } finally {
  96             asyncCloseEntityListenerReg.close();
  97             asyncCloseEntityListenerReg = null;
  98             serviceGroupMap.clear();
  99         }
 100
 101         LOG.info("Cluster Singleton Service stopped");
 102     }
 103
 104     @Override
 105     public synchronized ClusterSingletonServiceRegistration registerClusterSingletonService(
 106             final ClusterSingletonService service) {
 107         LOG.debug("Call registrationService {} method for ClusterSingletonService Provider {}", service, this);
 108
 109         final String serviceIdentifier = service.getIdentifier().getName();
 110         checkArgument(!Strings.isNullOrEmpty(serviceIdentifier),
 111             "ClusterSingletonService identifier may not be null nor empty");
 112
 113         final ClusterSingletonServiceGroup serviceGroup;
 114         final var existing = serviceGroupMap.get(serviceIdentifier);
 115         if (existing == null) {
 116             serviceGroup = createGroup(serviceIdentifier, new ArrayList<>(1));
 117             serviceGroupMap.put(serviceIdentifier, serviceGroup);
 118
 119             try {
 120                 initializeOrRemoveGroup(serviceGroup);
 121             } catch (CandidateAlreadyRegisteredException e) {
 122                 throw new IllegalArgumentException("Service group already registered", e);
 123             }
 124         } else {
 125             serviceGroup = existing;
 126         }
 127
 128         final var reg = new AbstractClusterSingletonServiceRegistration(service) {
 129             @Override
 130             protected void removeRegistration() {
 131                 // We need to bounce the unregistration through a ordered lock in order not to deal with asynchronous
 132                 // shutdown of the group and user registering it again.
 133                 EOSClusterSingletonServiceProvider.this.removeRegistration(serviceIdentifier, this);
 134             }
 135         };
 136
 137         serviceGroup.registerService(reg);
 138         return reg;
 139     }
 140
 141     private ClusterSingletonServiceGroup createGroup(final String serviceIdentifier,
 142             final List<ClusterSingletonServiceRegistration> services) {
 143         return new ClusterSingletonServiceGroupImpl(serviceIdentifier, entityOwnershipService,
 144             createEntity(SERVICE_ENTITY_TYPE, serviceIdentifier),
 145             createEntity(CLOSE_SERVICE_ENTITY_TYPE, serviceIdentifier), services);
 146     }
 147
 148     private void initializeOrRemoveGroup(final ClusterSingletonServiceGroup group)
 149             throws CandidateAlreadyRegisteredException {
 150         try {
 151             group.initialize();
 152         } catch (CandidateAlreadyRegisteredException e) {
 153             serviceGroupMap.remove(group.getIdentifier(), group);
 154             throw e;
 155         }
 156     }
 157
 158     private void removeRegistration(final String serviceIdentifier, final ClusterSingletonServiceRegistration reg) {
 159         final PlaceholderGroup placeHolder;
 160         final ListenableFuture<?> future;
 161         synchronized (this) {
 162             final var lookup = verifyNotNull(serviceGroupMap.get(serviceIdentifier));
 163             future = lookup.unregisterService(reg);
 164             if (future == null) {
 165                 return;
 166             }
 167
 168             // Close the group and replace it with a placeholder
 169             LOG.debug("Closing service group {}", serviceIdentifier);
 170             placeHolder = new PlaceholderGroup(lookup, future);
 171
 172             final String identifier = reg.getInstance().getIdentifier().getName();
 173             verify(serviceGroupMap.replace(identifier, lookup, placeHolder));
 174             LOG.debug("Replaced group {} with {}", serviceIdentifier, placeHolder);
 175
 176             lookup.closeClusterSingletonGroup();
 177         }
 178
 179         future.addListener(() -> finishShutdown(placeHolder), MoreExecutors.directExecutor());
 180     }
 181
 182     private synchronized void finishShutdown(final PlaceholderGroup placeHolder) {
 183         final var identifier = placeHolder.getIdentifier();
 184         LOG.debug("Service group {} closed", identifier);
 185
 186         final var services = placeHolder.getServices();
 187         if (services.isEmpty()) {
 188             // No services, we are all done
 189             if (serviceGroupMap.remove(identifier, placeHolder)) {
 190                 LOG.debug("Service group {} removed", placeHolder);
 191             } else {
 192                 LOG.debug("Service group {} superseded by {}", placeHolder, serviceGroupMap.get(identifier));
 193             }
 194             return;
 195         }
 196
 197         // Placeholder is being retired, we are reusing its services as the seed for the group.
 198         final var group = createGroup(identifier, services);
 199         verify(serviceGroupMap.replace(identifier, placeHolder, group));
 200         placeHolder.setSuccessor(group);
 201         LOG.debug("Service group upgraded from {} to {}", placeHolder, group);
 202
 203         try {
 204             initializeOrRemoveGroup(group);
 205         } catch (CandidateAlreadyRegisteredException e) {
 206             LOG.error("Failed to register delayed group {}, it will remain inoperational", identifier, e);
 207         }
 208     }
 209
 210     @Override
 211     public void ownershipChanged(final DOMEntity entity, final EntityOwnershipStateChange change,
 212             final boolean inJeopardy) {
 213         LOG.debug("Ownership change for ClusterSingletonService Provider on {} {} inJeopardy={}", entity, change,
 214             inJeopardy);
 215
 216         final var serviceIdentifier = getServiceIdentifierFromEntity(entity);
 217         final var serviceHolder = serviceGroupMap.get(serviceIdentifier);
 218         if (serviceHolder != null) {
 219             serviceHolder.ownershipChanged(entity, change, inJeopardy);
 220         } else {
 221             LOG.debug("ClusterSingletonServiceGroup was not found for serviceIdentifier {}", serviceIdentifier);
 222         }
 223     }
 224
 225     @VisibleForTesting
 226     static String getServiceIdentifierFromEntity(final DOMEntity entity) {
 227         final var yii = entity.getIdentifier();
 228         final var niiwp = (NodeIdentifierWithPredicates) yii.getLastPathArgument();
 229         return niiwp.values().iterator().next().toString();
 230     }
 231
 232     /**
 233      * Creates an extended {@link DOMEntity} instance.
 234      *
 235      * @param entityType the type of the entity
 236      * @param entityIdentifier the identifier of the entity
 237      * @return instance of Entity extended GenericEntity type
 238      */
 239     @VisibleForTesting
 240     static DOMEntity createEntity(final String entityType, final String entityIdentifier) {
 241         return new DOMEntity(entityType, entityIdentifier);
 242     }
 243 }