opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/FrontendClientMetadataBuilder.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.controller.cluster.datastore;
   9
  10 import static com.google.common.base.Verify.verify;
  11 import static java.util.Objects.requireNonNull;
  12
  13 import com.google.common.base.MoreObjects;
  14 import com.google.common.base.MoreObjects.ToStringHelper;
  15 import com.google.common.collect.Collections2;
  16 import com.google.common.collect.ImmutableList;
  17 import java.util.HashMap;
  18 import java.util.Map;
  19 import org.eclipse.jdt.annotation.NonNull;
  20 import org.opendaylight.controller.cluster.access.concepts.ClientIdentifier;
  21 import org.opendaylight.controller.cluster.access.concepts.LocalHistoryIdentifier;
  22 import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
  23 import org.opendaylight.controller.cluster.datastore.persisted.FrontendClientMetadata;
  24 import org.opendaylight.controller.cluster.datastore.persisted.FrontendHistoryMetadata;
  25 import org.opendaylight.controller.cluster.datastore.utils.ImmutableUnsignedLongSet;
  26 import org.opendaylight.controller.cluster.datastore.utils.MutableUnsignedLongSet;
  27 import org.opendaylight.yangtools.concepts.Identifiable;
  28 import org.slf4j.Logger;
  29 import org.slf4j.LoggerFactory;
  30
  31 /**
  32  * This class is NOT thread-safe.
  33  */
  34 // FIXME: sealed when we have JDK17+
  35 abstract class FrontendClientMetadataBuilder implements Identifiable<ClientIdentifier> {
  36     static final class Disabled extends FrontendClientMetadataBuilder {
  37         Disabled(final String shardName, final ClientIdentifier identifier) {
  38             super(shardName, identifier);
  39         }
  40
  41         @Override
  42         FrontendClientMetadata build() {
  43             return new FrontendClientMetadata(getIdentifier(), ImmutableUnsignedLongSet.of(), ImmutableList.of());
  44         }
  45
  46         @Override
  47         void onHistoryCreated(final LocalHistoryIdentifier historyId) {
  48             // No-op
  49         }
  50
  51         @Override
  52         void onHistoryClosed(final LocalHistoryIdentifier historyId) {
  53             // No-op
  54         }
  55
  56         @Override
  57         void onHistoryPurged(final LocalHistoryIdentifier historyId) {
  58             // No-op
  59         }
  60
  61         @Override
  62         void onTransactionAborted(final TransactionIdentifier txId) {
  63             // No-op
  64         }
  65
  66         @Override
  67         void onTransactionCommitted(final TransactionIdentifier txId) {
  68             // No-op
  69         }
  70
  71         @Override
  72         void onTransactionPurged(final TransactionIdentifier txId) {
  73             // No-op
  74         }
  75
  76         @Override
  77         void onTransactionsSkipped(final LocalHistoryIdentifier historyId, final ImmutableUnsignedLongSet txIds) {
  78             // No-op
  79         }
  80
  81         @Override
  82         LeaderFrontendState toLeaderState(final Shard shard) {
  83             return new LeaderFrontendState.Disabled(shard.persistenceId(), getIdentifier(), shard.getDataStore());
  84         }
  85     }
  86
  87     static final class Enabled extends FrontendClientMetadataBuilder {
  88         private final Map<LocalHistoryIdentifier, FrontendHistoryMetadataBuilder> currentHistories = new HashMap<>();
  89         private final MutableUnsignedLongSet purgedHistories;
  90         private final LocalHistoryIdentifier standaloneId;
  91
  92         Enabled(final String shardName, final ClientIdentifier identifier) {
  93             super(shardName, identifier);
  94
  95             purgedHistories = MutableUnsignedLongSet.of();
  96
  97             // History for stand-alone transactions is always present
  98             standaloneId = standaloneHistoryId();
  99             currentHistories.put(standaloneId, new FrontendHistoryMetadataBuilder(standaloneId));
 100         }
 101
 102         Enabled(final String shardName, final FrontendClientMetadata meta) {
 103             super(shardName, meta.getIdentifier());
 104
 105             purgedHistories = meta.getPurgedHistories().mutableCopy();
 106             for (FrontendHistoryMetadata h : meta.getCurrentHistories()) {
 107                 final FrontendHistoryMetadataBuilder b = new FrontendHistoryMetadataBuilder(getIdentifier(), h);
 108                 currentHistories.put(b.getIdentifier(), b);
 109             }
 110
 111             // Sanity check and recovery
 112             standaloneId = standaloneHistoryId();
 113             if (!currentHistories.containsKey(standaloneId)) {
 114                 LOG.warn("{}: Client {} recovered histories {} do not contain stand-alone history, attempting recovery",
 115                     shardName, getIdentifier(), currentHistories);
 116                 currentHistories.put(standaloneId, new FrontendHistoryMetadataBuilder(standaloneId));
 117             }
 118         }
 119
 120         @Override
 121         FrontendClientMetadata build() {
 122             return new FrontendClientMetadata(getIdentifier(), purgedHistories.immutableCopy(),
 123                 Collections2.transform(currentHistories.values(), FrontendHistoryMetadataBuilder::build));
 124         }
 125
 126         @Override
 127         void onHistoryCreated(final LocalHistoryIdentifier historyId) {
 128             final FrontendHistoryMetadataBuilder newMeta = new FrontendHistoryMetadataBuilder(historyId);
 129             final FrontendHistoryMetadataBuilder oldMeta = currentHistories.putIfAbsent(historyId, newMeta);
 130             if (oldMeta != null) {
 131                 // This should not be happening, warn about it
 132                 LOG.warn("{}: Reused local history {}", shardName(), historyId);
 133             } else {
 134                 LOG.debug("{}: Created local history {}", shardName(), historyId);
 135             }
 136         }
 137
 138         @Override
 139         void onHistoryClosed(final LocalHistoryIdentifier historyId) {
 140             final FrontendHistoryMetadataBuilder builder = currentHistories.get(historyId);
 141             if (builder != null) {
 142                 builder.onHistoryClosed();
 143                 LOG.debug("{}: Closed history {}", shardName(), historyId);
 144             } else {
 145                 LOG.warn("{}: Closed unknown history {}, ignoring", shardName(), historyId);
 146             }
 147         }
 148
 149         @Override
 150         void onHistoryPurged(final LocalHistoryIdentifier historyId) {
 151             final FrontendHistoryMetadataBuilder history = currentHistories.remove(historyId);
 152             final long historyBits = historyId.getHistoryId();
 153             if (history == null) {
 154                 if (!purgedHistories.contains(historyBits)) {
 155                     purgedHistories.add(historyBits);
 156                     LOG.warn("{}: Purging unknown history {}", shardName(), historyId);
 157                 } else {
 158                     LOG.warn("{}: Duplicate purge of history {}", shardName(), historyId);
 159                 }
 160             } else {
 161                 purgedHistories.add(historyBits);
 162                 LOG.debug("{}: Purged history {}", shardName(), historyId);
 163             }
 164         }
 165
 166         @Override
 167         void onTransactionAborted(final TransactionIdentifier txId) {
 168             final FrontendHistoryMetadataBuilder history = getHistory(txId);
 169             if (history != null) {
 170                 history.onTransactionAborted(txId);
 171                 LOG.debug("{}: Aborted transaction {}", shardName(), txId);
 172             } else {
 173                 LOG.warn("{}: Unknown history for aborted transaction {}, ignoring", shardName(), txId);
 174             }
 175         }
 176
 177         @Override
 178         void onTransactionCommitted(final TransactionIdentifier txId) {
 179             final FrontendHistoryMetadataBuilder history = getHistory(txId);
 180             if (history != null) {
 181                 history.onTransactionCommitted(txId);
 182                 LOG.debug("{}: Committed transaction {}", shardName(), txId);
 183             } else {
 184                 LOG.warn("{}: Unknown history for commited transaction {}, ignoring", shardName(), txId);
 185             }
 186         }
 187
 188         @Override
 189         void onTransactionPurged(final TransactionIdentifier txId) {
 190             final FrontendHistoryMetadataBuilder history = getHistory(txId);
 191             if (history != null) {
 192                 history.onTransactionPurged(txId);
 193                 LOG.debug("{}: Purged transaction {}", shardName(), txId);
 194             } else {
 195                 LOG.warn("{}: Unknown history for purged transaction {}, ignoring", shardName(), txId);
 196             }
 197         }
 198
 199         @Override
 200         void onTransactionsSkipped(final LocalHistoryIdentifier historyId, final ImmutableUnsignedLongSet txIds) {
 201             final FrontendHistoryMetadataBuilder history = getHistory(historyId);
 202             if (history != null) {
 203                 history.onTransactionsSkipped(txIds);
 204                 LOG.debug("{}: History {} skipped transactions {}", shardName(), historyId, txIds);
 205             } else {
 206                 LOG.warn("{}: Unknown history {} for skipped transactions, ignoring", shardName(), historyId);
 207             }
 208         }
 209
 210         @Override
 211         LeaderFrontendState toLeaderState(final Shard shard) {
 212             // Note: we have to make sure to *copy* all current state and not leak any views, otherwise leader/follower
 213             //       interactions would get intertwined leading to inconsistencies.
 214             final Map<LocalHistoryIdentifier, LocalFrontendHistory> histories = new HashMap<>();
 215             for (FrontendHistoryMetadataBuilder e : currentHistories.values()) {
 216                 if (e.getIdentifier().getHistoryId() != 0) {
 217                     final AbstractFrontendHistory state = e.toLeaderState(shard);
 218                     verify(state instanceof LocalFrontendHistory, "Unexpected state %s", state);
 219                     histories.put(e.getIdentifier(), (LocalFrontendHistory) state);
 220                 }
 221             }
 222
 223             final AbstractFrontendHistory singleHistory;
 224             final FrontendHistoryMetadataBuilder singleHistoryMeta = currentHistories.get(
 225                 new LocalHistoryIdentifier(getIdentifier(), 0));
 226             if (singleHistoryMeta == null) {
 227                 final ShardDataTree tree = shard.getDataStore();
 228                 singleHistory = StandaloneFrontendHistory.create(shard.persistenceId(), getIdentifier(), tree);
 229             } else {
 230                 singleHistory = singleHistoryMeta.toLeaderState(shard);
 231             }
 232
 233             return new LeaderFrontendState.Enabled(shard.persistenceId(), getIdentifier(), shard.getDataStore(),
 234                 purgedHistories.mutableCopy(), singleHistory, histories);
 235         }
 236
 237         @Override
 238         ToStringHelper addToStringAttributes(final ToStringHelper helper) {
 239             return super.addToStringAttributes(helper).add("current", currentHistories).add("purged", purgedHistories);
 240         }
 241
 242         private FrontendHistoryMetadataBuilder getHistory(final TransactionIdentifier txId) {
 243             return getHistory(txId.getHistoryId());
 244         }
 245
 246         private FrontendHistoryMetadataBuilder getHistory(final LocalHistoryIdentifier historyId) {
 247             final LocalHistoryIdentifier local;
 248             if (historyId.getHistoryId() == 0 && historyId.getCookie() != 0) {
 249                 // We are pre-creating the history for free-standing transactions with a zero cookie, hence our lookup
 250                 // needs to account for that.
 251                 LOG.debug("{}: looking up {} instead of {}", shardName(), standaloneId, historyId);
 252                 local = standaloneId;
 253             } else {
 254                 local = historyId;
 255             }
 256
 257             return currentHistories.get(local);
 258         }
 259
 260         private LocalHistoryIdentifier standaloneHistoryId() {
 261             return new LocalHistoryIdentifier(getIdentifier(), 0);
 262         }
 263     }
 264
 265     private static final Logger LOG = LoggerFactory.getLogger(FrontendClientMetadataBuilder.class);
 266
 267     private final @NonNull ClientIdentifier identifier;
 268     private final @NonNull String shardName;
 269
 270     FrontendClientMetadataBuilder(final String shardName, final ClientIdentifier identifier) {
 271         this.shardName = requireNonNull(shardName);
 272         this.identifier = requireNonNull(identifier);
 273     }
 274
 275     static FrontendClientMetadataBuilder of(final String shardName, final FrontendClientMetadata meta) {
 276         // Completely empty histories imply disabled state, as otherwise we'd have a record of the single history --
 277         // either purged or active
 278         return meta.getCurrentHistories().isEmpty() && meta.getPurgedHistories().isEmpty()
 279             ? new Disabled(shardName, meta.getIdentifier()) : new Enabled(shardName, meta);
 280     }
 281
 282     @Override
 283     public final ClientIdentifier getIdentifier() {
 284         return identifier;
 285     }
 286
 287     final String shardName() {
 288         return shardName;
 289     }
 290
 291     abstract FrontendClientMetadata build();
 292
 293     abstract void onHistoryCreated(LocalHistoryIdentifier historyId);
 294
 295     abstract void onHistoryClosed(LocalHistoryIdentifier historyId);
 296
 297     abstract void onHistoryPurged(LocalHistoryIdentifier historyId);
 298
 299     abstract void onTransactionAborted(TransactionIdentifier txId);
 300
 301     abstract void onTransactionCommitted(TransactionIdentifier txId);
 302
 303     abstract void onTransactionPurged(TransactionIdentifier txId);
 304
 305     abstract void onTransactionsSkipped(LocalHistoryIdentifier historyId, ImmutableUnsignedLongSet txIds);
 306
 307     /**
 308      * Transform frontend metadata for a particular client into its {@link LeaderFrontendState} counterpart.
 309      *
 310      * @param shard parent shard
 311      * @return Leader frontend state
 312      */
 313     abstract @NonNull LeaderFrontendState toLeaderState(@NonNull Shard shard);
 314
 315     @Override
 316     public final String toString() {
 317         return addToStringAttributes(MoreObjects.toStringHelper(this)).toString();
 318     }
 319
 320     ToStringHelper addToStringAttributes(final ToStringHelper helper) {
 321         return helper.add("identifier", identifier);
 322     }
 323 }