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