esctl/pkg/cfg/cluster.go

/*
Copyright © 2026 Thomas von Dein

This program is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.

This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
GNU General Public License for more details.

You should have received a copy of the GNU General Public License
along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
package cfg

import (
	"crypto/tls"
	"errors"
	"fmt"
	"net"
	"net/http"
	"os"
	"strings"
	"syscall"
	"time"

	"github.com/elastic/elastic-transport-go/v8/elastictransport"
	"github.com/elastic/go-elasticsearch/v9"
	"golang.org/x/term"
	"gopkg.in/yaml.v3"
)

// used in general config struct
type Cluster struct {
	Name, Uri, User, Pass, Token string
	client                       *elasticsearch.TypedClient
	Default, DebugHTTP           bool
}

// used just for writing back to the config file
type ClusterConfig struct {
	Uri, User, Pass, Token string
	Default                bool
}

// to write the config, we avoid all other config settings
type WriteConfig struct {
	Clusters map[string]*ClusterConfig
}

func (cluster *Cluster) SetClient(client *elasticsearch.TypedClient) {
	cluster.client = client
}

// return the  go-elasticsearch client  object but before  doing that,
// check if we need to tune auth
func (cluster *Cluster) ES() *elasticsearch.TypedClient {
	if cluster.client == nil {
		fmt.Println("no current cluster, use 'esctl cluster switch <name>' to set one")
		os.Exit(1)
	}

	if err := cluster.CheckAuth(); err != nil {
		fmt.Printf("Error: %s", err)
		os.Exit(1)
	}

	return cluster.client
}

// add authentication to es client, if not yet done
func (cluster *Cluster) CheckAuth() error {
	if cluster.Pass == "" && cluster.User != "" && cluster.Token == "" && cluster.Default {
		// no token - user is set, but no password.

		// check if the env var is set
		pass := os.Getenv("ES_PASS")
		if pass != "" {
			cluster.Pass = pass
		} else {
			// k, try interactively
			fmt.Fprintf(os.Stderr, "Enter password for elasticsearch user %s@%s: ", cluster.User, cluster.Name)

			pass, err := term.ReadPassword(syscall.Stdin)
			if err != nil {
				return err
			}

			passwd := strings.TrimSpace(string(pass))
			if passwd == "" {
				return errors.New("password empty")
			}

			cluster.Pass = string(pass)

			fmt.Println()
		}

		opts := cluster.getDefaultOptions()
		opts = append(opts, elasticsearch.WithBasicAuth(cluster.User, cluster.Pass))

		es, err := elasticsearch.NewTyped(opts...)
		if err != nil {
			return fmt.Errorf("failed to setup elasticsearch connection: %w", err)
		}

		cluster.SetClient(es)
	}

	return nil
}

// set Default=true for the given cluster in the config (if exists)
func (conf *Config) SwitchCluster(name string) error {
	_, exists := conf.Clusters[name]
	if !exists {
		return errors.New("no cluster with that name configured")
	}

	cfg := WriteConfig{Clusters: map[string]*ClusterConfig{}}

	for clustername, cluster := range conf.Clusters {
		cfg.Clusters[clustername] = &ClusterConfig{
			Uri:     cluster.Uri,
			User:    cluster.User,
			Pass:    cluster.Pass,
			Token:   cluster.Token,
			Default: false,
		}

		if clustername == name {
			cfg.Clusters[clustername].Default = true
		}
	}

	raw, err := yaml.Marshal(cfg)
	if err != nil {
		return fmt.Errorf("failed to marshal cluster config: %w", err)
	}

	outfile := getDefaultPath()
	if conf.ConfigFile != "" {
		outfile = conf.ConfigFile
	}

	if err := os.WriteFile(outfile, raw, 0600); err != nil {
		return err
	}

	return nil
}

// We do NOT  use go-elasticsearch to check  for cluster reachability,
// because  at this  stage,  auth  may not  have  been configured.  So
// instead we just connect to  the cluster using plan net/tcp
func (cluster *Cluster) IsReachable() (bool, error) {
	timeout := 500 * time.Millisecond

	url := strings.TrimPrefix(strings.TrimPrefix(cluster.Uri, "https://"), "http://")

	host := strings.Split(url, "/")

	if !strings.Contains(host[0], ":") {
		host[0] += ":443"
	}

	conn, err := net.DialTimeout("tcp", host[0], timeout)
	if err != nil {
		return false, err
	}

	return true, conn.Close()
}

func (cluster *Cluster) getDefaultOptions() []elasticsearch.Option {
	// These headers are  not needed with ES 9, but with  ES 8, we set
	// them here  so every API call uses it.  The only exception being
	// the api repl, which does it on its own.
	headers := http.Header{}
	headers.Add("Content-Type", "application/json")
	headers.Add("Accept", "application/json")

	return []elasticsearch.Option{
		elasticsearch.WithAddresses(cluster.Uri),
		elasticsearch.WithTransportOptions(
			cluster.getTransport(),
			elastictransport.WithHeader(headers),
		),
		elasticsearch.WithCompatibilityMode(),
	}
}

func (cluster *Cluster) getTransport() elastictransport.Option {
	transport := &http.Transport{
		TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
	}

	if cluster.DebugHTTP {
		return elastictransport.WithTransport(
			&DebugTransport{Transport: transport},
		)
	}

	return elastictransport.WithTransport(
		&CompatibilityTransport{Transport: transport},
	)
}

func (conf *Config) SetupES() error {
	for name, cluster := range conf.Clusters {
		cluster.Name = name
		cluster.DebugHTTP = conf.DebugHTTP

		opts := cluster.getDefaultOptions()

		switch {
		case cluster.Pass != "" && cluster.User != "":
			opts = append(opts, elasticsearch.WithBasicAuth(cluster.User, cluster.Pass))
		case cluster.Token != "":
			opts = append(opts, elasticsearch.WithAPIKey(cluster.Token))
		}

		es, err := elasticsearch.NewTyped(opts...)
		if err != nil {
			return fmt.Errorf("failed to setup elasticsearch connection: %w", err)
		}

		cluster.SetClient(es)
	}

	return nil
}