mirror of
https://codeberg.org/scip/esctl.git
synced 2026-08-24 04:54:18 +02:00
ask for pass if missing or use env pass, add token support (#60)
This commit is contained in:
T. von Dein
@@ -17,27 +17,33 @@ along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
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 <name>' 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)
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -17,7 +17,6 @@ along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
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
|
||||
|
||||
Reference in New Issue
Block a user