replicate/mdsal-replicate-netty/src/main/java/org/opendaylight/mdsal/replicate/netty/SinkSingletonService.java

   1 /*
   2  * Copyright (c) 2020 PANTHEON.tech, s.r.o. 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.replicate.netty;
   9
  10 import static java.util.Objects.requireNonNull;
  11
  12 import com.google.common.util.concurrent.FutureCallback;
  13 import com.google.common.util.concurrent.ListenableFuture;
  14 import io.netty.bootstrap.Bootstrap;
  15 import io.netty.buffer.ByteBuf;
  16 import io.netty.buffer.ByteBufOutputStream;
  17 import io.netty.buffer.Unpooled;
  18 import io.netty.channel.Channel;
  19 import io.netty.channel.ChannelFuture;
  20 import io.netty.channel.ChannelFutureListener;
  21 import io.netty.channel.ChannelInitializer;
  22 import io.netty.channel.ChannelOption;
  23 import io.netty.channel.socket.SocketChannel;
  24 import io.netty.handler.timeout.IdleStateHandler;
  25 import java.io.IOException;
  26 import java.net.InetSocketAddress;
  27 import java.time.Duration;
  28 import java.util.concurrent.ScheduledExecutorService;
  29 import java.util.concurrent.TimeUnit;
  30 import org.checkerframework.checker.lock.qual.GuardedBy;
  31 import org.checkerframework.checker.lock.qual.Holding;
  32 import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
  33 import org.opendaylight.mdsal.dom.api.DOMDataBroker;
  34 import org.opendaylight.mdsal.dom.api.DOMDataTreeIdentifier;
  35 import org.opendaylight.mdsal.singleton.common.api.ClusterSingletonService;
  36 import org.opendaylight.mdsal.singleton.common.api.ServiceGroupIdentifier;
  37 import org.opendaylight.yangtools.util.concurrent.FluentFutures;
  38 import org.opendaylight.yangtools.yang.common.Empty;
  39 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  40 import org.opendaylight.yangtools.yang.data.codec.binfmt.NormalizedNodeDataOutput;
  41 import org.opendaylight.yangtools.yang.data.codec.binfmt.NormalizedNodeStreamVersion;
  42 import org.slf4j.Logger;
  43 import org.slf4j.LoggerFactory;
  44
  45 final class SinkSingletonService extends ChannelInitializer<SocketChannel> implements ClusterSingletonService {
  46     private static final Logger LOG = LoggerFactory.getLogger(SinkSingletonService.class);
  47     private static final ServiceGroupIdentifier SGID = new ServiceGroupIdentifier(SinkSingletonService.class.getName());
  48     // TODO: allow different trees?
  49     private static final DOMDataTreeIdentifier TREE = DOMDataTreeIdentifier.of(LogicalDatastoreType.CONFIGURATION,
  50         YangInstanceIdentifier.of());
  51     private static long CHANNEL_CLOSE_TIMEOUT_S = 10;
  52     private static final ByteBuf TREE_REQUEST;
  53
  54     static {
  55         try {
  56             TREE_REQUEST = Unpooled.unreleasableBuffer(requestTree(TREE));
  57         } catch (IOException e) {
  58             throw new ExceptionInInitializerError(e);
  59         }
  60     }
  61
  62     private final BootstrapSupport bootstrapSupport;
  63     private final DOMDataBroker dataBroker;
  64     private final InetSocketAddress sourceAddress;
  65     private final Duration reconnectDelay;
  66     private final int maxMissedKeepalives;
  67     private final Duration keepaliveInterval;
  68
  69     @GuardedBy("this")
  70     private ChannelFuture futureChannel;
  71     private boolean closingInstance;
  72     private Bootstrap bs;
  73
  74     SinkSingletonService(final BootstrapSupport bootstrapSupport, final DOMDataBroker dataBroker,
  75             final InetSocketAddress sourceAddress, final Duration reconnectDelay, final Duration keepaliveInterval,
  76             final int maxMissedKeepalives) {
  77         this.bootstrapSupport = requireNonNull(bootstrapSupport);
  78         this.dataBroker = requireNonNull(dataBroker);
  79         this.sourceAddress = requireNonNull(sourceAddress);
  80         this.reconnectDelay = requireNonNull(reconnectDelay);
  81         this.keepaliveInterval = requireNonNull(keepaliveInterval);
  82         this.maxMissedKeepalives = maxMissedKeepalives;
  83         LOG.info("Replication sink from {} waiting for cluster-wide mastership", sourceAddress);
  84     }
  85
  86     @Override
  87     public ServiceGroupIdentifier getIdentifier() {
  88         return SGID;
  89     }
  90
  91     @Override
  92     public synchronized void instantiateServiceInstance() {
  93         LOG.info("Replication sink started with source {}", sourceAddress);
  94         bs = bootstrapSupport.newBootstrap();
  95         doConnect();
  96     }
  97
  98     @Holding("this")
  99     private void doConnect() {
 100         LOG.info("Connecting to Source");
 101         final ScheduledExecutorService group = bs.config().group();
 102
 103         futureChannel = bs
 104             .option(ChannelOption.SO_KEEPALIVE, true)
 105             .handler(this)
 106             .connect(sourceAddress, null);
 107         futureChannel.addListener((ChannelFutureListener) future -> channelResolved(future, group));
 108     }
 109
 110     @Override
 111     public synchronized ListenableFuture<?> closeServiceInstance() {
 112         closingInstance = true;
 113         if (futureChannel == null) {
 114             return FluentFutures.immediateNullFluentFuture();
 115         }
 116
 117         return FluentFutures.immediateBooleanFluentFuture(disconnect());
 118     }
 119
 120     private synchronized void reconnect() {
 121         disconnect();
 122         doConnect();
 123     }
 124
 125     private synchronized boolean disconnect() {
 126         boolean shutdownSuccess = true;
 127         final Channel channel = futureChannel.channel();
 128         if (channel != null && channel.isActive()) {
 129             try {
 130                 // close the resulting channel. Even when this triggers the closeFuture, it won't try to reconnect since
 131                 // the closingInstance flag is set
 132                 channel.close().await(CHANNEL_CLOSE_TIMEOUT_S, TimeUnit.SECONDS);
 133             } catch (InterruptedException e) {
 134                 LOG.error("The channel didn't close properly within {} seconds", CHANNEL_CLOSE_TIMEOUT_S);
 135                 shutdownSuccess = false;
 136             }
 137         }
 138         shutdownSuccess &= futureChannel.cancel(true);
 139         futureChannel = null;
 140         return shutdownSuccess;
 141     }
 142
 143     @Override
 144     protected void initChannel(final SocketChannel ch) {
 145         final var txChain = dataBroker.createMergingTransactionChain();
 146
 147         ch.pipeline()
 148             .addLast("frameDecoder", new MessageFrameDecoder())
 149             .addLast("idleStateHandler", new IdleStateHandler(
 150                 keepaliveInterval.toNanos() * maxMissedKeepalives, 0, 0, TimeUnit.NANOSECONDS))
 151             .addLast("keepaliveHandler", new SinkKeepaliveHandler())
 152             .addLast("requestHandler", new SinkRequestHandler(TREE, txChain))
 153             .addLast("frameEncoder", MessageFrameEncoder.INSTANCE);
 154
 155         txChain.addCallback(new FutureCallback<>() {
 156             @Override
 157             public void onSuccess(final Empty result) {
 158                 LOG.info("Transaction chain for channel {} completed", ch);
 159             }
 160
 161             @Override
 162             public void onFailure(final Throwable cause) {
 163                 LOG.error("Transaction chain for channel {} failed", ch, cause);
 164                 ch.close();
 165             }
 166         });
 167     }
 168
 169     private synchronized void channelResolved(final ChannelFuture completedFuture,
 170             final ScheduledExecutorService group) {
 171         if (futureChannel != null && futureChannel.channel() == completedFuture.channel()) {
 172             if (completedFuture.isSuccess()) {
 173                 final Channel ch = completedFuture.channel();
 174                 LOG.info("Channel {} established", ch);
 175                 ch.closeFuture().addListener((ChannelFutureListener) future -> channelClosed(future, group));
 176                 ch.writeAndFlush(TREE_REQUEST);
 177             } else {
 178                 LOG.info("Failed to connect to source {}, reconnecting in {}", sourceAddress,
 179                     reconnectDelay.getSeconds(), completedFuture.cause());
 180                 group.schedule(() -> {
 181                     reconnect();
 182                 }, reconnectDelay.toNanos(), TimeUnit.NANOSECONDS);
 183             }
 184         }
 185     }
 186
 187     private synchronized void channelClosed(final ChannelFuture completedFuture, final ScheduledExecutorService group) {
 188         if (futureChannel != null && futureChannel.channel() == completedFuture.channel() && !closingInstance) {
 189             LOG.info("Channel {} lost connection to source {}, reconnecting in {}", completedFuture.channel(),
 190                 sourceAddress, reconnectDelay.getSeconds());
 191             group.schedule(this::reconnect, reconnectDelay.toNanos(), TimeUnit.NANOSECONDS);
 192         }
 193     }
 194
 195     private static ByteBuf requestTree(final DOMDataTreeIdentifier tree) throws IOException {
 196         final ByteBuf ret = Unpooled.buffer();
 197
 198         try (ByteBufOutputStream stream = new ByteBufOutputStream(ret)) {
 199             stream.writeByte(Constants.MSG_SUBSCRIBE_REQ);
 200             try (NormalizedNodeDataOutput output = NormalizedNodeStreamVersion.current().newDataOutput(stream)) {
 201                 tree.datastore().writeTo(output);
 202                 output.writeYangInstanceIdentifier(tree.path());
 203             }
 204         }
 205
 206         return ret;
 207     }
 208 }