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