diff --git a/README.md b/README.md
index 8560cd3..367969f 100644
--- a/README.md
+++ b/README.md
@@ -113,6 +113,7 @@ Configure `esctl` with environment variables:
- `ES_URI`: elasticsearch uri
- `ES_USER`: username
- `ES_PASS`: password
+- `ES_TOKEN`: API token, instead of user+password
Or create a config file such as this:
@@ -121,11 +122,10 @@ clusters:
foobar:
uri: https://es.foo.bar:9200/
user: elastic
- pass: 123456
+ pass: ******
other:
uri: https://myes.foo:9200/
- user: elastic
- pass: asdasdasd
+ token: ******
```
and specify it with `-c configfile`. You may also put clusters into a
diff --git a/cmd/cluster.go b/cmd/cluster.go
index 91d646c..fc3cbd6 100644
--- a/cmd/cluster.go
+++ b/cmd/cluster.go
@@ -61,12 +61,6 @@ func ClusterStatus(conf *cfg.Config) *cli.Command {
Aliases: []string{"s"},
Flags: []cli.Flag{
- &cli.BoolFlag{
- Name: "all",
- Usage: "show status of all clusters",
- Destination: &conf.All,
- Aliases: []string{"a"},
- },
&cli.BoolFlag{
Name: "verbose",
Usage: "include verbose statistics",
diff --git a/pkg/cfg/cluster.go b/pkg/cfg/cluster.go
index ca0ae3a..49ef88c 100644
--- a/pkg/cfg/cluster.go
+++ b/pkg/cfg/cluster.go
@@ -17,27 +17,33 @@ along with this program. If not, see .
package cfg
import (
+ "context"
+ "crypto/tls"
"errors"
"fmt"
"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 {
- Uri, User, Pass string
- client *elasticsearch.TypedClient
- Default bool
+ 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 string
- Default bool
+ Uri, User, Pass, Token string
+ Default bool
}
// to write the config, we avoid all other config settings
@@ -45,17 +51,97 @@ type WriteConfig struct {
Clusters map[string]*ClusterConfig
}
+func (cluster *Cluster) SetClient(client *elasticsearch.TypedClient) {
+ cluster.client = client
+}
+
+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(transport)
+}
+
+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),
+ ),
+ }
+}
+
+// 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 ' to set one")
os.Exit(1)
}
+ if err := cluster.CheckAuth(); err != nil {
+ fmt.Printf("Error: %s", err)
+ os.Exit(1)
+ }
+
return cluster.client
}
-func (cluster *Cluster) SetClient(client *elasticsearch.TypedClient) {
- cluster.client = 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.Printf("Enter password for elasticsearch user %s@%s: ", cluster.User, cluster.Name)
+ pass, err := term.ReadPassword(int(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)
@@ -72,6 +158,7 @@ func (conf *Config) SwitchCluster(name string) error {
Uri: cluster.Uri,
User: cluster.User,
Pass: cluster.Pass,
+ Token: cluster.Token,
Default: false,
}
@@ -97,23 +184,61 @@ func (conf *Config) SwitchCluster(name string) error {
return nil
}
-func (conf *Config) SetupES() error {
- // 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")
+// 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/http, ignore
+// HTTP response status and return true if we could just reach ith
+func (cluster *Cluster) IsReachable() (bool, error) {
+ ctx, cancel := context.WithTimeout(
+ context.Background(),
+ time.Duration(500)*time.Millisecond)
+ defer cancel()
- for _, cluster := range conf.Clusters {
- es, err := elasticsearch.NewTyped(
- elasticsearch.WithAddresses(cluster.Uri),
- elasticsearch.WithBasicAuth(cluster.User, cluster.Pass),
- elasticsearch.WithTransportOptions(
- conf.getTransport(),
- elastictransport.WithHeader(headers),
- ),
- )
+ req, err := http.NewRequestWithContext(
+ ctx,
+ "GET",
+ cluster.Uri,
+ nil,
+ )
+
+ if err != nil {
+ return false, err
+ }
+
+ client := &http.Client{Transport: &http.Transport{
+ TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
+ }}
+
+ resp, err := client.Do(req)
+
+ if err != nil {
+ return false, err
+ }
+
+ if resp != nil {
+ // at this stage we do not care if the elasticsearch cluster
+ // accepts our request or if it's misconfigured in some way
+ return true, nil
+ }
+
+ return false, nil
+}
+
+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)
diff --git a/pkg/cfg/config.go b/pkg/cfg/config.go
index 5d0c8af..ac61757 100644
--- a/pkg/cfg/config.go
+++ b/pkg/cfg/config.go
@@ -28,7 +28,7 @@ import (
)
const (
- Version string = `v0.0.23`
+ Version string = `v0.0.24`
)
var (
@@ -213,18 +213,17 @@ func (conf *Config) PrintDebug() {
func (conf *Config) LoadEnv() error {
cluster := Cluster{
- Uri: os.Getenv("ES_URI"),
- User: os.Getenv("ES_USER"),
- Pass: os.Getenv("ES_PASS"),
+ Uri: os.Getenv("ES_URI"),
+ User: os.Getenv("ES_USER"),
+ Pass: os.Getenv("ES_PASS"),
+ Token: os.Getenv("ES_TOKEN"),
}
switch {
case cluster.Uri == "":
return errors.New("ES_URI unset")
- case cluster.User == "":
- return errors.New("ES_USER unset")
- case cluster.Pass == "":
- return errors.New("ES_PASS unset")
+ case cluster.User == "" || cluster.Token == "":
+ return errors.New("ES_USER and ES_TOKEN unset")
}
conf.Clusters["default"] = &cluster
diff --git a/pkg/cfg/transport.go b/pkg/cfg/transport.go
index 386edfb..3eb17b5 100644
--- a/pkg/cfg/transport.go
+++ b/pkg/cfg/transport.go
@@ -18,13 +18,10 @@ package cfg
import (
"bytes"
- "crypto/tls"
"encoding/json"
"fmt"
"log/slog"
"net/http"
-
- "github.com/elastic/elastic-transport-go/v8/elastictransport"
)
// used to print uri, path and body of a request made by the go-client
@@ -62,17 +59,3 @@ func (t *DebugTransport) RoundTrip(req *http.Request) (*http.Response, error) {
return t.Transport.RoundTrip(req)
}
-
-func (conf *Config) getTransport() elastictransport.Option {
- transport := &http.Transport{
- TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
- }
-
- if conf.DebugHTTP {
- return elastictransport.WithTransport(
- &DebugTransport{Transport: transport},
- )
- }
-
- return elastictransport.WithTransport(transport)
-}
diff --git a/pkg/es/cluster.go b/pkg/es/cluster.go
index 66ba216..571e0f9 100644
--- a/pkg/es/cluster.go
+++ b/pkg/es/cluster.go
@@ -17,7 +17,6 @@ along with this program. If not, see .
package es
import (
- "context"
"fmt"
"log/slog"
"strings"
@@ -38,32 +37,33 @@ import (
type ClusterIndices map[string]map[string]*types.IndicesRecord
func ClusterList(conf *cfg.Config) error {
- table := printer.NewTable(conf, 4, len(conf.Clusters))
+ table := printer.NewTable(conf, 5, len(conf.Clusters))
- table.Addheaders("cluster", "uri", "reachable", "current")
+ table.Addheaders("cluster", "uri", "reachable", "current", "error")
idx := 0
for name, cluster := range conf.Clusters {
reachable := "no"
current := "no"
+ errmsg := ""
- _, err := cluster.ES().Cluster.Health().
- Do(context.Background())
+ online, err := cluster.IsReachable()
- if err == nil {
+ if online {
reachable = printer.Colorize(conf, "green", "reachable")
}
if cluster.Default {
current = printer.Colorize(conf, "green", "yes")
- if err != nil {
+ if !online {
reachable = printer.Colorize(conf, "red", "no")
+ errmsg = err.Error()
}
}
- table.Entries[idx] = []string{name, cluster.Uri, reachable, current}
+ table.Entries[idx] = []string{name, cluster.Uri, reachable, current, errmsg}
idx++
}
@@ -79,134 +79,120 @@ func ClusterList(conf *cfg.Config) error {
// We're using goroutines here to parallelize API requests, since we
// have to do 3 of'em for each cluster. This speeds things up.
func ClusterStatus(conf *cfg.Config) error {
- clusters := []string{}
gocount := 5
if conf.Verbose {
gocount++
}
- if conf.All {
- for key := range conf.Clusters {
- clusters = append(clusters, key)
- }
- } else {
- clusters = []string{"default"}
+ es := conf.DefaultCluster.ES()
+
+ responses := make(chan apiResponse, gocount)
+ wg := &sync.WaitGroup{}
+
+ wg.Add(gocount)
+ go getApiData(es, wg, responses, "health")
+ go getApiData(es, wg, responses, "info")
+ go getApiData(es, wg, responses, "ccrstats")
+ go getApiData(es, wg, responses, "indices")
+ go getApiData(es, wg, responses, "tasks")
+
+ if conf.Verbose {
+ go getApiData(es, wg, responses, "stats")
}
- for _, cluster := range clusters {
- es := conf.DefaultCluster.ES()
- if cluster != "default" {
- es = conf.Clusters[cluster].ES()
+ wg.Wait()
+
+ var clusterhealth *health.Response
+ var info *info.Response
+ var ccrstats *stats.Response
+ var clusterstats *clusterstats.Response
+ var indexstats *indices.Response
+ var taskstatus *tasks.Response
+
+ for i := 0; i < gocount; i++ {
+ r := <-responses
+
+ if r.error != nil {
+ return r.error
}
- responses := make(chan apiResponse, gocount)
- wg := &sync.WaitGroup{}
-
- wg.Add(gocount)
- go getApiData(es, wg, responses, "health")
- go getApiData(es, wg, responses, "info")
- go getApiData(es, wg, responses, "ccrstats")
- go getApiData(es, wg, responses, "indices")
- go getApiData(es, wg, responses, "tasks")
-
- if conf.Verbose {
- go getApiData(es, wg, responses, "stats")
+ switch r.which {
+ case ResponseHealth:
+ clusterhealth = r.health
+ case ResponseCcr:
+ ccrstats = r.ccr
+ case ResponseInfo:
+ info = r.info
+ case ResponseStats:
+ clusterstats = r.stats
+ case ResponseIndices:
+ indexstats = r.indices
+ case ResponseTasks:
+ taskstatus = r.tasks
}
+ }
- wg.Wait()
+ slog.Debug("ES result", "cluster health", clusterhealth)
- var clusterhealth *health.Response
- var info *info.Response
- var ccrstats *stats.Response
- var clusterstats *clusterstats.Response
- var indexstats *indices.Response
- var taskstatus *tasks.Response
+ isleader := len(ccrstats.AutoFollowStats.AutoFollowedClusters) == 0
- for i := 0; i < gocount; i++ {
- r := <-responses
+ ccrfollowing := ""
+ if len(ccrstats.AutoFollowStats.AutoFollowedClusters) > 0 {
+ // is following another cluster
+ ccrfollowing = fmt.Sprintf("%s (%d/%d)",
+ ccrstats.AutoFollowStats.AutoFollowedClusters[0].ClusterName,
+ ccrstats.AutoFollowStats.NumberOfSuccessfulFollowIndices,
+ ccrstats.AutoFollowStats.NumberOfFailedFollowIndices,
+ )
+ }
- if r.error != nil {
- return r.error
- }
-
- switch r.which {
- case ResponseHealth:
- clusterhealth = r.health
- case ResponseCcr:
- ccrstats = r.ccr
- case ResponseInfo:
- info = r.info
- case ResponseStats:
- clusterstats = r.stats
- case ResponseIndices:
- indexstats = r.indices
- case ResponseTasks:
- taskstatus = r.tasks
- }
+ // look for red indices, if any
+ redindices := 0
+ for _, index := range *indexstats {
+ if *index.Health == "red" {
+ redindices++
}
+ }
- slog.Debug("ES result", "cluster health", clusterhealth)
-
- isleader := len(ccrstats.AutoFollowStats.AutoFollowedClusters) == 0
-
- ccrfollowing := ""
- if len(ccrstats.AutoFollowStats.AutoFollowedClusters) > 0 {
- // is following another cluster
- ccrfollowing = fmt.Sprintf("%s (%d/%d)",
- ccrstats.AutoFollowStats.AutoFollowedClusters[0].ClusterName,
- ccrstats.AutoFollowStats.NumberOfSuccessfulFollowIndices,
- ccrstats.AutoFollowStats.NumberOfFailedFollowIndices,
- )
+ // look for long running tasks
+ longtasks := 0
+ for _, task := range *taskstatus {
+ if strings.Contains(*task.RunningTime, "d") {
+ longtasks++
}
+ }
- // look for red indices, if any
- redindices := 0
- for _, index := range *indexstats {
- if *index.Health == "red" {
- redindices++
- }
- }
+ table := printer.NewTable(conf, 2, 7)
+ table.Addheaders(conf.DefaultCluster.Name, "status")
- // look for long running tasks
- longtasks := 0
- for _, task := range *taskstatus {
- if strings.Contains(*task.RunningTime, "d") {
- longtasks++
- }
- }
+ table.Entries = [][]string{
+ {"Cluster Name", clusterhealth.ClusterName},
+ {"ES Status", printer.Colorize(conf, clusterhealth.Status.Name, clusterhealth.Status.Name)},
+ {"ES Version", info.Version.Int},
+ {"Is Leader", fmt.Sprintf("%t", isleader)},
+ {"Active Shards", fmt.Sprintf("%d", clusterhealth.ActiveShards)},
+ {"Active Primary Shards", fmt.Sprintf("%d", clusterhealth.ActivePrimaryShards)},
+ {"Unassigned Shards", fmt.Sprintf("%d", clusterhealth.UnassignedShards)},
+ {"Unassigned Primary Shards", fmt.Sprintf("%d", clusterhealth.UnassignedPrimaryShards)},
+ {"Pending Tasks", fmt.Sprintf("%d", clusterhealth.NumberOfPendingTasks)},
+ {"Nodes", fmt.Sprintf("%d", clusterhealth.NumberOfNodes)},
+ {"Red Indices", fmt.Sprintf("%d", redindices)},
+ {"Long Running Tasks", fmt.Sprintf("%d", longtasks)},
+ }
- table := printer.NewTable(conf, 2, 7)
- table.Addheaders(cluster, "status")
+ if !isleader {
+ table.Entries = append(table.Entries, [][]string{
+ {"AutoFollow (success/failed indices)", ccrfollowing},
+ {"Followed Indices", fmt.Sprintf("%d", len(ccrstats.FollowStats.Indices))},
+ }...)
+ }
- table.Entries = [][]string{
- {"Cluster Name", clusterhealth.ClusterName},
- {"ES Status", printer.Colorize(conf, clusterhealth.Status.Name, clusterhealth.Status.Name)},
- {"ES Version", info.Version.Int},
- {"Is Leader", fmt.Sprintf("%t", isleader)},
- {"Active Shards", fmt.Sprintf("%d", clusterhealth.ActiveShards)},
- {"Active Primary Shards", fmt.Sprintf("%d", clusterhealth.ActivePrimaryShards)},
- {"Unassigned Shards", fmt.Sprintf("%d", clusterhealth.UnassignedShards)},
- {"Unassigned Primary Shards", fmt.Sprintf("%d", clusterhealth.UnassignedPrimaryShards)},
- {"Pending Tasks", fmt.Sprintf("%d", clusterhealth.NumberOfPendingTasks)},
- {"Nodes", fmt.Sprintf("%d", clusterhealth.NumberOfNodes)},
- {"Red Indices", fmt.Sprintf("%d", redindices)},
- {"Long Running Tasks", fmt.Sprintf("%d", longtasks)},
- }
+ if conf.Verbose {
+ table = gatherClusterStats(conf, clusterstats, table)
+ }
- if !isleader {
- table.Entries = append(table.Entries, [][]string{
- {"AutoFollow (success/failed indices)", ccrfollowing},
- {"Followed Indices", fmt.Sprintf("%d", len(ccrstats.FollowStats.Indices))},
- }...)
- }
-
- if conf.Verbose {
- table = gatherClusterStats(conf, clusterstats, table)
- }
-
- if err := table.Print(); err != nil {
- return err
- }
+ if err := table.Print(); err != nil {
+ return err
}
return nil