package cluster // import "github.com/docker/docker/daemon/cluster"

import (
	"context"
	"fmt"
	"path/filepath"
	"strings"
	"sync"
	"time"

	"github.com/containerd/log"
	types "github.com/docker/docker/api/types/swarm"
	"github.com/docker/docker/daemon/cluster/convert"
	"github.com/docker/docker/daemon/cluster/executor/container"
	lncluster "github.com/docker/docker/libnetwork/cluster"
	"github.com/docker/docker/libnetwork/cnmallocator"
	swarmapi "github.com/moby/swarmkit/v2/api"
	"github.com/moby/swarmkit/v2/manager/allocator/networkallocator"
	swarmnode "github.com/moby/swarmkit/v2/node"
	"github.com/pkg/errors"
	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
)

// nodeRunner implements a manager for continuously running swarmkit node, restarting them with backoff delays if needed.
type nodeRunner struct {
	nodeState
	mu             sync.RWMutex
	done           chan struct{} // closed when swarmNode exits
	ready          chan struct{} // closed when swarmNode becomes active
	reconnectDelay time.Duration
	config         nodeStartConfig

	repeatedRun     bool
	cancelReconnect func()
	stopping        bool
	cluster         *Cluster // only for accessing config helpers, never call any methods. TODO: change to config struct
}

// nodeStartConfig holds configuration needed to start a new node. Exported
// fields of this structure are saved to disk in json. Unexported fields
// contain data that shouldn't be persisted between daemon reloads.
type nodeStartConfig struct {
	// LocalAddr is this machine's local IP or hostname, if specified.
	LocalAddr string
	// RemoteAddr is the address that was given to "swarm join". It is used
	// to find LocalAddr if necessary.
	RemoteAddr string
	// ListenAddr is the address we bind to, including a port.
	ListenAddr string
	// AdvertiseAddr is the address other nodes should connect to,
	// including a port.
	AdvertiseAddr string
	// DataPathAddr is the address that has to be used for the data path
	DataPathAddr string
	// DefaultAddressPool contains list of subnets
	DefaultAddressPool []string
	// SubnetSize contains subnet size of DefaultAddressPool
	SubnetSize uint32
	// DataPathPort contains Data path port (VXLAN UDP port) number that is used for data traffic.
	DataPathPort uint32
	// JoinInProgress is set to true if a join operation has started, but
	// not completed yet.
	JoinInProgress bool

	joinAddr        string
	forceNewCluster bool
	joinToken       string
	lockKey         []byte
	autolock        bool
	availability    types.NodeAvailability
}

func (n *nodeRunner) Ready() chan error {
	c := make(chan error, 1)
	n.mu.RLock()
	ready, done := n.ready, n.done
	n.mu.RUnlock()
	go func() {
		select {
		case <-ready:
		case <-done:
		}
		select {
		case <-ready:
		default:
			n.mu.RLock()
			c <- n.err
			n.mu.RUnlock()
		}
		close(c)
	}()
	return c
}

func (n *nodeRunner) Start(conf nodeStartConfig) error {
	n.mu.Lock()
	defer n.mu.Unlock()

	n.reconnectDelay = initialReconnectDelay

	return n.start(conf)
}

func (n *nodeRunner) start(conf nodeStartConfig) error {
	var control string
	if isWindows {
		control = `\\.\pipe\` + controlSocket
	} else {
		control = filepath.Join(n.cluster.runtimeRoot, controlSocket)
	}

	joinAddr := conf.joinAddr
	if joinAddr == "" && conf.JoinInProgress {
		// We must have been restarted while trying to join a cluster.
		// Continue trying to join instead of forming our own cluster.
		joinAddr = conf.RemoteAddr
	}

	// Hostname is not set here. Instead, it is obtained from
	// the node description that is reported periodically
	swarmnodeConfig := swarmnode.Config{
		ForceNewCluster:    conf.forceNewCluster,
		ListenControlAPI:   control,
		ListenRemoteAPI:    conf.ListenAddr,
		AdvertiseRemoteAPI: conf.AdvertiseAddr,
		NetworkConfig: &networkallocator.Config{
			DefaultAddrPool: conf.DefaultAddressPool,
			SubnetSize:      conf.SubnetSize,
			VXLANUDPPort:    conf.DataPathPort,
		},
		JoinAddr:  joinAddr,
		StateDir:  n.cluster.root,
		JoinToken: conf.joinToken,
		Executor: container.NewExecutor(
			n.cluster.config.Backend,
			n.cluster.config.PluginBackend,
			n.cluster.config.ImageBackend,
			n.cluster.config.VolumeBackend,
		),
		HeartbeatTick: n.cluster.config.RaftHeartbeatTick,
		// Recommended value in etcd/raft is 10 x (HeartbeatTick).
		// Lower values were seen to have caused instability because of
		// frequent leader elections when running on flakey networks.
		ElectionTick:     n.cluster.config.RaftElectionTick,
		UnlockKey:        conf.lockKey,
		AutoLockManagers: conf.autolock,
		PluginGetter:     convert.SwarmPluginGetter(n.cluster.config.Backend.PluginGetter()),
		NetworkProvider:  cnmallocator.NewProvider(n.cluster.config.Backend.PluginGetter()),
	}
	if conf.availability != "" {
		avail, ok := swarmapi.NodeSpec_Availability_value[strings.ToUpper(string(conf.availability))]
		if !ok {
			return fmt.Errorf("invalid Availability: %q", conf.availability)
		}
		swarmnodeConfig.Availability = swarmapi.NodeSpec_Availability(avail)
	}
	node, err := swarmnode.New(&swarmnodeConfig)
	if err != nil {
		return err
	}
	if err := node.Start(context.Background()); err != nil {
		return err
	}

	n.done = make(chan struct{})
	n.ready = make(chan struct{})
	n.swarmNode = node
	if conf.joinAddr != "" {
		conf.JoinInProgress = true
	}
	n.config = conf
	savePersistentState(n.cluster.root, conf)

	ctx, cancel := context.WithCancel(context.Background())

	go func() {
		n.handleNodeExit(node)
		cancel()
	}()

	go n.handleReadyEvent(ctx, node, n.ready)
	go n.handleControlSocketChange(ctx, node)

	return nil
}

func (n *nodeRunner) handleControlSocketChange(ctx context.Context, node *swarmnode.Node) {
	for conn := range node.ListenControlSocket(ctx) {
		n.mu.Lock()
		if n.grpcConn != conn {
			if conn == nil {
				n.controlClient = nil
				n.logsClient = nil
			} else {
				n.controlClient = swarmapi.NewControlClient(conn)
				n.logsClient = swarmapi.NewLogsClient(conn)
				// push store changes to daemon
				go n.watchClusterEvents(ctx, conn)
			}
		}
		n.grpcConn = conn
		n.mu.Unlock()
		n.cluster.SendClusterEvent(lncluster.EventSocketChange)
	}
}

func (n *nodeRunner) watchClusterEvents(ctx context.Context, conn *grpc.ClientConn) {
	client := swarmapi.NewWatchClient(conn)
	watch, err := client.Watch(ctx, &swarmapi.WatchRequest{
		Entries: []*swarmapi.WatchRequest_WatchEntry{
			{
				Kind:   "node",
				Action: swarmapi.WatchActionKindCreate | swarmapi.WatchActionKindUpdate | swarmapi.WatchActionKindRemove,
			},
			{
				Kind:   "service",
				Action: swarmapi.WatchActionKindCreate | swarmapi.WatchActionKindUpdate | swarmapi.WatchActionKindRemove,
			},
			{
				Kind:   "network",
				Action: swarmapi.WatchActionKindCreate | swarmapi.WatchActionKindUpdate | swarmapi.WatchActionKindRemove,
			},
			{
				Kind:   "secret",
				Action: swarmapi.WatchActionKindCreate | swarmapi.WatchActionKindUpdate | swarmapi.WatchActionKindRemove,
			},
			{
				Kind:   "config",
				Action: swarmapi.WatchActionKindCreate | swarmapi.WatchActionKindUpdate | swarmapi.WatchActionKindRemove,
			},
		},
		IncludeOldObject: true,
	})
	if err != nil {
		log.G(ctx).WithError(err).Error("failed to watch cluster store")
		return
	}
	for {
		msg, err := watch.Recv()
		if err != nil {
			// store watch is broken
			errStatus, ok := status.FromError(err)
			if !ok || errStatus.Code() != codes.Canceled {
				log.G(ctx).WithError(err).Error("failed to receive changes from store watch API")
			}
			return
		}
		select {
		case <-ctx.Done():
			return
		case n.cluster.watchStream <- msg:
		}
	}
}

func (n *nodeRunner) handleReadyEvent(ctx context.Context, node *swarmnode.Node, ready chan struct{}) {
	select {
	case <-node.Ready():
		n.mu.Lock()
		n.err = nil
		if n.config.JoinInProgress {
			n.config.JoinInProgress = false
			savePersistentState(n.cluster.root, n.config)
		}
		n.mu.Unlock()
		close(ready)
	case <-ctx.Done():
	}
	n.cluster.SendClusterEvent(lncluster.EventNodeReady)
}

func (n *nodeRunner) handleNodeExit(node *swarmnode.Node) {
	err := detectLockedError(node.Err(context.Background()))
	if err != nil {
		log.G(context.TODO()).Errorf("cluster exited with error: %v", err)
	}
	n.mu.Lock()
	n.swarmNode = nil
	n.err = err
	close(n.done)
	select {
	case <-n.ready:
		// there is a case where a node can be promoted to manager while
		// another node is leaving the cluster. the node being promoted, by
		// random chance, picks the IP of the node being demoted as the one it
		// tries to connect to. in this case, the promotion will fail, and the
		// whole swarm Node object packs it in.
		//
		// when the Node object is relaunched by this code, because it has
		// joinAddr in the config, it attempts again to connect to the same
		// no-longer-manager node, and crashes again. this continues forever.
		//
		// to avoid this case, in this block, we remove JoinAddr from the
		// config.
		n.config.joinAddr = ""
		n.enableReconnectWatcher()
	default:
		if n.repeatedRun {
			n.enableReconnectWatcher()
		}
	}
	n.repeatedRun = true
	n.mu.Unlock()
}

// Stop stops the current swarm node if it is running.
func (n *nodeRunner) Stop() error {
	n.mu.Lock()
	if n.cancelReconnect != nil { // between restarts
		n.cancelReconnect()
		n.cancelReconnect = nil
	}
	if n.swarmNode == nil {
		// even though the swarm node is nil we still may need
		// to send a node leave event to perform any cleanup required.
		if n.cluster != nil {
			n.cluster.SendClusterEvent(lncluster.EventNodeLeave)
		}
		n.mu.Unlock()
		return nil
	}
	n.stopping = true
	ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
	defer cancel()
	n.mu.Unlock()
	if err := n.swarmNode.Stop(ctx); err != nil && !strings.Contains(err.Error(), "context canceled") {
		return err
	}
	n.cluster.SendClusterEvent(lncluster.EventNodeLeave)
	<-n.done
	return nil
}

func (n *nodeRunner) State() nodeState {
	if n == nil {
		return nodeState{status: types.LocalNodeStateInactive}
	}
	n.mu.RLock()
	defer n.mu.RUnlock()

	ns := n.nodeState

	if ns.err != nil || n.cancelReconnect != nil {
		if errors.Is(ns.err, errSwarmLocked) {
			ns.status = types.LocalNodeStateLocked
		} else {
			ns.status = types.LocalNodeStateError
		}
	} else {
		select {
		case <-n.ready:
			ns.status = types.LocalNodeStateActive
		default:
			ns.status = types.LocalNodeStatePending
		}
	}

	return ns
}

func (n *nodeRunner) enableReconnectWatcher() {
	if n.stopping {
		return
	}
	n.reconnectDelay *= 2
	if n.reconnectDelay > maxReconnectDelay {
		n.reconnectDelay = maxReconnectDelay
	}
	log.G(context.TODO()).Warnf("Restarting swarm in %.2f seconds", n.reconnectDelay.Seconds())
	delayCtx, cancel := context.WithTimeout(context.Background(), n.reconnectDelay)
	n.cancelReconnect = cancel

	go func() {
		<-delayCtx.Done()
		if delayCtx.Err() != context.DeadlineExceeded {
			return
		}
		n.mu.Lock()
		defer n.mu.Unlock()
		if n.stopping {
			return
		}

		if err := n.start(n.config); err != nil {
			n.err = err
		}
	}()
}

// nodeState represents information about the current state of the cluster and
// provides access to the grpc clients.
type nodeState struct {
	swarmNode       *swarmnode.Node
	grpcConn        *grpc.ClientConn
	controlClient   swarmapi.ControlClient
	logsClient      swarmapi.LogsClient
	status          types.LocalNodeState
	actualLocalAddr string
	err             error
}

// IsActiveManager returns true if node is a manager ready to accept control requests. It is safe to access the client properties if this returns true.
func (ns nodeState) IsActiveManager() bool {
	return ns.controlClient != nil
}

// IsManager returns true if node is a manager.
func (ns nodeState) IsManager() bool {
	return ns.swarmNode != nil && ns.swarmNode.Manager() != nil
}

// NodeID returns node's ID or empty string if node is inactive.
func (ns nodeState) NodeID() string {
	if ns.swarmNode != nil {
		return ns.swarmNode.NodeID()
	}
	return ""
}