singleton-service/mdsal-singleton-dom-impl/src/main/java/org/opendaylight/mdsal/singleton/dom/impl/AbstractClusterSingletonServiceProviderImpl.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.FutureCallback;
  18 import com.google.common.util.concurrent.Futures;
  19 import com.google.common.util.concurrent.ListenableFuture;
  20 import com.google.common.util.concurrent.MoreExecutors;
  21 import java.util.ArrayList;
  22 import java.util.List;
  23 import java.util.concurrent.ConcurrentHashMap;
  24 import java.util.concurrent.ConcurrentMap;
  25 import org.eclipse.jdt.annotation.NonNull;
  26 import org.opendaylight.mdsal.eos.common.api.CandidateAlreadyRegisteredException;
  27 import org.opendaylight.mdsal.eos.common.api.EntityOwnershipStateChange;
  28 import org.opendaylight.mdsal.eos.dom.api.DOMEntity;
  29 import org.opendaylight.mdsal.eos.dom.api.DOMEntityOwnershipListener;
  30 import org.opendaylight.mdsal.eos.dom.api.DOMEntityOwnershipService;
  31 import org.opendaylight.mdsal.singleton.common.api.ClusterSingletonService;
  32 import org.opendaylight.mdsal.singleton.common.api.ClusterSingletonServiceProvider;
  33 import org.opendaylight.mdsal.singleton.common.api.ClusterSingletonServiceRegistration;
  34 import org.opendaylight.yangtools.concepts.Registration;
  35 import org.slf4j.Logger;
  36 import org.slf4j.LoggerFactory;
  37
  38 /**
  39  * Abstract class {@link AbstractClusterSingletonServiceProviderImpl} represents implementations of
  40  * {@link ClusterSingletonServiceProvider} and it implements {@link DOMEntityOwnershipListener} for providing
  41  * OwnershipChange for all registered {@link ClusterSingletonServiceGroup} entity candidate.
  42  */
  43 public abstract class AbstractClusterSingletonServiceProviderImpl
  44         implements ClusterSingletonServiceProvider, DOMEntityOwnershipListener {
  45     private static final Logger LOG = LoggerFactory.getLogger(AbstractClusterSingletonServiceProviderImpl.class);
  46
  47     @VisibleForTesting
  48     static final @NonNull String SERVICE_ENTITY_TYPE = "org.opendaylight.mdsal.ServiceEntityType";
  49     @VisibleForTesting
  50     static final @NonNull String CLOSE_SERVICE_ENTITY_TYPE = "org.opendaylight.mdsal.AsyncServiceCloseEntityType";
  51
  52     private final ConcurrentMap<String, ClusterSingletonServiceGroup> serviceGroupMap = new ConcurrentHashMap<>();
  53     private final DOMEntityOwnershipService entityOwnershipService;
  54
  55     /* EOS Entity Listeners Registration */
  56     private Registration serviceEntityListenerReg;
  57     private Registration asyncCloseEntityListenerReg;
  58
  59     /**
  60      * Class constructor.
  61      *
  62      * @param entityOwnershipService relevant EOS
  63      */
  64     protected AbstractClusterSingletonServiceProviderImpl(
  65             final @NonNull DOMEntityOwnershipService entityOwnershipService) {
  66         this.entityOwnershipService = requireNonNull(entityOwnershipService);
  67     }
  68
  69     /**
  70      * This method must be called once on startup to initialize this provider.
  71      */
  72     public final void initializeProvider() {
  73         LOG.debug("Initialization method for ClusterSingletonService Provider {}", this);
  74         serviceEntityListenerReg = registerListener(SERVICE_ENTITY_TYPE, entityOwnershipService);
  75         asyncCloseEntityListenerReg = registerListener(CLOSE_SERVICE_ENTITY_TYPE, entityOwnershipService);
  76     }
  77
  78     @Override
  79     public final synchronized ClusterSingletonServiceRegistration registerClusterSingletonService(
  80             final ClusterSingletonService service) {
  81         LOG.debug("Call registrationService {} method for ClusterSingletonService Provider {}", service, this);
  82
  83         final String serviceIdentifier = service.getIdentifier().getName();
  84         checkArgument(!Strings.isNullOrEmpty(serviceIdentifier),
  85             "ClusterSingletonService identifier may not be null nor empty");
  86
  87         final ClusterSingletonServiceGroup serviceGroup;
  88         final var existing = serviceGroupMap.get(serviceIdentifier);
  89         if (existing == null) {
  90             serviceGroup = createGroup(serviceIdentifier, new ArrayList<>(1));
  91             serviceGroupMap.put(serviceIdentifier, serviceGroup);
  92
  93             try {
  94                 initializeOrRemoveGroup(serviceGroup);
  95             } catch (CandidateAlreadyRegisteredException e) {
  96                 throw new IllegalArgumentException("Service group already registered", e);
  97             }
  98         } else {
  99             serviceGroup = existing;
 100         }
 101
 102         final var reg =  new AbstractClusterSingletonServiceRegistration(service) {
 103             @Override
 104             protected void removeRegistration() {
 105                 // We need to bounce the unregistration through a ordered lock in order not to deal with asynchronous
 106                 // shutdown of the group and user registering it again.
 107                 AbstractClusterSingletonServiceProviderImpl.this.removeRegistration(serviceIdentifier, this);
 108             }
 109         };
 110
 111         serviceGroup.registerService(reg);
 112         return reg;
 113     }
 114
 115     private ClusterSingletonServiceGroup createGroup(final String serviceIdentifier,
 116             final List<ClusterSingletonServiceRegistration> services) {
 117         return new ClusterSingletonServiceGroupImpl(serviceIdentifier, entityOwnershipService,
 118             createEntity(SERVICE_ENTITY_TYPE, serviceIdentifier),
 119             createEntity(CLOSE_SERVICE_ENTITY_TYPE, serviceIdentifier), services);
 120     }
 121
 122     private void initializeOrRemoveGroup(final ClusterSingletonServiceGroup group)
 123             throws CandidateAlreadyRegisteredException {
 124         try {
 125             group.initialize();
 126         } catch (CandidateAlreadyRegisteredException e) {
 127             serviceGroupMap.remove(group.getIdentifier(), group);
 128             throw e;
 129         }
 130     }
 131
 132     void removeRegistration(final String serviceIdentifier, final ClusterSingletonServiceRegistration reg) {
 133         final PlaceholderGroup placeHolder;
 134         final ListenableFuture<?> future;
 135         synchronized (this) {
 136             final var lookup = verifyNotNull(serviceGroupMap.get(serviceIdentifier));
 137             future = lookup.unregisterService(reg);
 138             if (future == null) {
 139                 return;
 140             }
 141
 142             // Close the group and replace it with a placeholder
 143             LOG.debug("Closing service group {}", serviceIdentifier);
 144             placeHolder = new PlaceholderGroup(lookup, future);
 145
 146             final String identifier = reg.getInstance().getIdentifier().getName();
 147             verify(serviceGroupMap.replace(identifier, lookup, placeHolder));
 148             LOG.debug("Replaced group {} with {}", serviceIdentifier, placeHolder);
 149
 150             lookup.closeClusterSingletonGroup();
 151         }
 152
 153         future.addListener(() -> finishShutdown(placeHolder), MoreExecutors.directExecutor());
 154     }
 155
 156     synchronized void finishShutdown(final PlaceholderGroup placeHolder) {
 157         final var identifier = placeHolder.getIdentifier();
 158         LOG.debug("Service group {} closed", identifier);
 159
 160         final var services = placeHolder.getServices();
 161         if (services.isEmpty()) {
 162             // No services, we are all done
 163             if (serviceGroupMap.remove(identifier, placeHolder)) {
 164                 LOG.debug("Service group {} removed", placeHolder);
 165             } else {
 166                 LOG.debug("Service group {} superseded by {}", placeHolder, serviceGroupMap.get(identifier));
 167             }
 168             return;
 169         }
 170
 171         // Placeholder is being retired, we are reusing its services as the seed for the group.
 172         final var group = createGroup(identifier, services);
 173         verify(serviceGroupMap.replace(identifier, placeHolder, group));
 174         placeHolder.setSuccessor(group);
 175         LOG.debug("Service group upgraded from {} to {}", placeHolder, group);
 176
 177         try {
 178             initializeOrRemoveGroup(group);
 179         } catch (CandidateAlreadyRegisteredException e) {
 180             LOG.error("Failed to register delayed group {}, it will remain inoperational", identifier, e);
 181         }
 182     }
 183
 184     @Override
 185     public final void close() {
 186         LOG.debug("Close method for ClusterSingletonService Provider {}", this);
 187
 188         if (serviceEntityListenerReg != null) {
 189             serviceEntityListenerReg.close();
 190             serviceEntityListenerReg = null;
 191         }
 192
 193         final var futures = serviceGroupMap.values().stream()
 194             .map(ClusterSingletonServiceGroup::closeClusterSingletonGroup)
 195             .toList();
 196         final var future = Futures.allAsList(futures);
 197         Futures.addCallback(future, new FutureCallback<>() {
 198             @Override
 199             public void onSuccess(final List<Object> result) {
 200                 cleanup();
 201             }
 202
 203             @Override
 204             public void onFailure(final Throwable throwable) {
 205                 LOG.warn("Unexpected problem by closing ClusterSingletonServiceProvider {}",
 206                     AbstractClusterSingletonServiceProviderImpl.this, throwable);
 207                 cleanup();
 208             }
 209         }, MoreExecutors.directExecutor());
 210     }
 211
 212     @Override
 213     public final void ownershipChanged(final DOMEntity entity, final EntityOwnershipStateChange change,
 214             final boolean inJeopardy) {
 215         LOG.debug("Ownership change for ClusterSingletonService Provider on {} {} inJeopardy={}", entity, change,
 216             inJeopardy);
 217         final var serviceIdentifier = getServiceIdentifierFromEntity(entity);
 218         final var serviceHolder = serviceGroupMap.get(serviceIdentifier);
 219         if (serviceHolder != null) {
 220             serviceHolder.ownershipChanged(entity, change, inJeopardy);
 221         } else {
 222             LOG.debug("ClusterSingletonServiceGroup was not found for serviceIdentifier {}", serviceIdentifier);
 223         }
 224     }
 225
 226     /**
 227      * Method implementation registers the listener.
 228      *
 229      * @param entityType the type of the entity
 230      * @param eos - EOS type
 231      * @return a {@link Registration}
 232      */
 233     protected abstract Registration registerListener(String entityType, DOMEntityOwnershipService eos);
 234
 235     /**
 236      * Creates an extended {@link DOMEntity} instance.
 237      *
 238      * @param entityType the type of the entity
 239      * @param entityIdentifier the identifier of the entity
 240      * @return instance of Entity extended GenericEntity type
 241      */
 242     @VisibleForTesting
 243     static final DOMEntity createEntity(final String entityType, final String entityIdentifier) {
 244         return new DOMEntity(entityType, entityIdentifier);
 245     }
 246
 247     /**
 248      * Method is responsible for parsing ServiceGroupIdentifier from E entity.
 249      *
 250      * @param entity instance of GenericEntity type
 251      * @return ServiceGroupIdentifier parsed from entity key value.
 252      */
 253     protected abstract String getServiceIdentifierFromEntity(DOMEntity entity);
 254
 255     /**
 256      * Method is called async. from close method in end of Provider lifecycle.
 257      */
 258     final void cleanup() {
 259         LOG.debug("Final cleaning ClusterSingletonServiceProvider {}", this);
 260         if (asyncCloseEntityListenerReg != null) {
 261             asyncCloseEntityListenerReg.close();
 262             asyncCloseEntityListenerReg = null;
 263         }
 264         serviceGroupMap.clear();
 265     }
 266 }