func (c *cluster) Containers() container.ContainersInterface {
	return c.containers
}
Containers implement types.Resource interface.
package etcd
import (
	"context"
	"fmt"
	"sort"
	"strings"
	"time"
	clientv3 "go.etcd.io/etcd/client/v3"
	"sigs.k8s.io/yaml"
	"github.com/flexkube/libflexkube/internal/util"
	"github.com/flexkube/libflexkube/pkg/container"
	containertypes "github.com/flexkube/libflexkube/pkg/container/types"
	"github.com/flexkube/libflexkube/pkg/defaults"
	"github.com/flexkube/libflexkube/pkg/host"
	"github.com/flexkube/libflexkube/pkg/host/transport/ssh"
	"github.com/flexkube/libflexkube/pkg/pki"
	"github.com/flexkube/libflexkube/pkg/types"
)
const defaultDialTimeout = 5 * time.Second
type Cluster struct {
	Image string `json:"image,omitempty"`
	SSH *ssh.Config `json:"ssh,omitempty"`
	CACertificate string `json:"caCertificate,omitempty"`
	PeerCertAllowedCN string `json:"peerCertAllowedCN,omitempty"`
	Members map[string]MemberConfig `json:"members,omitempty"`
	PKI *pki.PKI `json:"pki,omitempty"`
	State container.ContainersState `json:"state,omitempty"`
	ExtraMounts []containertypes.Mount `json:"extraMounts,omitempty"`
}
type cluster struct {
	containers container.ContainersInterface
	members    map[string]Member
}
func (c *Cluster) propagateMember(i string, m *MemberConfig) {
	initialClusterArr := []string{}
	peerCertAllowedCNArr := []string{}
	for n, m := range c.Members {
		name := util.PickString(m.Name, n)
		initialClusterArr = append(initialClusterArr, fmt.Sprintf("%s=https://%s:2380", name, m.PeerAddress))
		peerCertAllowedCNArr = append(peerCertAllowedCNArr, name)
	}
	sort.Strings(initialClusterArr)
	sort.Strings(peerCertAllowedCNArr)
	m.Name = util.PickString(m.Name, i)
	m.Image = util.PickString(m.Image, c.Image, defaults.EtcdImage)
	m.InitialCluster = util.PickString(m.InitialCluster, strings.Join(initialClusterArr, ","))
	m.PeerCertAllowedCN = util.PickString(m.PeerCertAllowedCN, c.PeerCertAllowedCN)
	m.CACertificate = util.PickString(m.CACertificate, c.CACertificate)
	if len(m.ExtraMounts) == 0 {
		m.ExtraMounts = c.ExtraMounts
	}
	if c.PKI != nil && c.PKI.Etcd != nil {
		e := c.PKI.Etcd
		m.CACertificate = util.PickString(m.CACertificate, c.CACertificate, string(e.CA.X509Certificate))
		if c, ok := e.PeerCertificates[m.Name]; ok {
			m.PeerCertificate = util.PickString(m.PeerCertificate, string(c.X509Certificate))
			m.PeerKey = util.PickString(m.PeerKey, string(c.PrivateKey))
		}
		if c, ok := e.ServerCertificates[m.Name]; ok {
			m.ServerCertificate = util.PickString(m.ServerCertificate, string(c.X509Certificate))
			m.ServerKey = util.PickString(m.ServerKey, string(c.PrivateKey))
		}
	}
	m.ServerAddress = util.PickString(m.ServerAddress, m.PeerAddress)
	m.Host = host.BuildConfig(m.Host, host.Host{
		SSHConfig: c.SSH,
	})
	if len(c.State) == 0 {
		m.NewCluster = true
	}
}
func (c *Cluster) New() (types.Resource, error) {
	if err := c.Validate(); err != nil {
		return nil, fmt.Errorf("validating cluster configuration: %w", err)
	}
	cc := container.Containers{
		PreviousState: c.State,
		DesiredState:  container.ContainersState{},
	}
	cluster := &cluster{
		members: map[string]Member{},
	}
	for n, m := range c.Members {
		m := m
		c.propagateMember(n, &m)
		mem, _ := m.New()                          
		hcc, _ := mem.ToHostConfiguredContainer()  
		cc.DesiredState[n] = hcc
		cluster.members[n] = mem
	}
	co, _ := cc.New()  
	cluster.containers = co
	return cluster, nil
}
func (c *Cluster) Validate() error {
	if len(c.Members) == 0 && len(c.State) == 0 {
		return fmt.Errorf("at least one member must be defined when state is empty")
	}
	var errors util.ValidateErrors
	if c.CACertificate != "" {
		caCert := &pki.Certificate{
			X509Certificate: types.Certificate(c.CACertificate),
		}
		if _, err := caCert.DecodeX509Certificate(); err != nil {
			errors = append(errors, fmt.Errorf("parsing CA certificate: %w", err))
		}
	}
	cc := container.Containers{
		PreviousState: c.State,
		DesiredState:  container.ContainersState{},
	}
	for n, m := range c.Members {
		m := m
		c.propagateMember(n, &m)
		mem, err := m.New()
		if err != nil {
			errors = append(errors, fmt.Errorf("validating member %q: %w", n, err))
			continue
		}
		hcc, err := mem.ToHostConfiguredContainer()
		if err != nil {
			errors = append(errors, fmt.Errorf("validating member %q container: %w", n, err))
			continue
		}
		cc.DesiredState[n] = hcc
	}
	if _, err := cc.New(); err != nil {
		errors = append(errors, fmt.Errorf("validating containers object: %w", err))
	}
	return errors.Return()
}
func FromYaml(c []byte) (types.Resource, error) {
	return types.ResourceFromYaml(c, &Cluster{})
}
func (c *cluster) StateToYaml() ([]byte, error) {
	return yaml.Marshal(Cluster{State: c.containers.ToExported().PreviousState})
}
func (c *cluster) CheckCurrentState() error {
	if err := c.containers.CheckCurrentState(); err != nil {
		return fmt.Errorf("checking current state of etcd cluster: %w", err)
	}
	return nil
}
func (c *cluster) getExistingEndpoints() []string {
	endpoints := []string{}
	for i, m := range c.members {
		if _, ok := c.containers.ToExported().PreviousState[i]; !ok {
			continue
		}
		endpoints = append(endpoints, fmt.Sprintf("%s:2379", m.peerAddress()))
	}
	return endpoints
}
func (c *cluster) firstMember() (Member, error) {
	for i := range c.members {
		return c.members[i], nil
	}
	return nil, fmt.Errorf("no members defined")
}
func (c *cluster) getClient() (etcdClient, error) {
	m, err := c.firstMember()
	if err != nil {
		return nil, fmt.Errorf("getting member object: %w", err)
	}
	endpoints, err := m.forwardEndpoints(c.getExistingEndpoints())
	if err != nil {
		return nil, fmt.Errorf("forwarding endpoints: %w", err)
	}
	return m.getEtcdClient(endpoints)
}
type etcdClient interface {
	MemberList(context context.Context) (*clientv3.MemberListResponse, error)
	MemberAdd(context context.Context, peerURLs []string) (*clientv3.MemberAddResponse, error)
	MemberRemove(context context.Context, id uint64) (*clientv3.MemberRemoveResponse, error)
	Close() error
}
func (c *cluster) membersToRemove() []string {
	m := []string{}
	e := c.containers.ToExported()
	for i := range e.PreviousState {
		if _, ok := e.DesiredState[i]; !ok {
			m = append(m, i)
		}
	}
	return m
}
func (c *cluster) membersToAdd() []string {
	m := []string{}
	e := c.containers.ToExported()
	for i := range e.DesiredState {
		if _, ok := e.PreviousState[i]; !ok {
			m = append(m, i)
		}
	}
	return m
}
func (c *cluster) updateMembers(cli etcdClient) error {
	for _, name := range c.membersToRemove() {
		m := &member{
			config: &MemberConfig{
				Name: name,
			},
		}
		if err := m.remove(cli); err != nil {
			return fmt.Errorf("removing member: %w", err)
		}
	}
	for _, m := range c.membersToAdd() {
		if err := c.members[m].add(cli); err != nil {
			return fmt.Errorf("adding member: %w", err)
		}
	}
	return nil
}
func (c *cluster) Deploy() error {
	e := c.containers.ToExported()
	if len(e.PreviousState) != 0 && len(e.DesiredState) != 0 {
		cli, err := c.getClient()
		if err != nil {
			return fmt.Errorf("getting etcd client: %w", err)
		}
		if err := c.updateMembers(cli); err != nil {
			return fmt.Errorf("updating members before deploying: %w", err)
		}
		if err := cli.Close(); err != nil {
			return fmt.Errorf("closing etcd client: %w", err)
		}
	}
	return c.containers.Deploy()
}