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