Compare commits

...

8 Commits

Author SHA1 Message Date
4f230b5f02 satisfy linter 2026-07-09 13:38:18 +02:00
b7ac5733f1 fix #85 2026-07-09 13:11:35 +02:00
4557fcca7b fix #84 2026-07-09 13:11:20 +02:00
1e9e419e9b fix crash in ByteSize conversion 2026-07-08 15:07:01 +02:00
T. von Dein
c128e32ca1 add more important node stats (#82) 2026-07-08 15:03:31 +02:00
2969522913 rename var 2026-07-08 15:00:21 +02:00
51480e6b35 add 'index du <index>' 2026-07-08 14:58:07 +02:00
99bf703997 add table.AddRowLate() to add rows after sorting 2026-07-08 14:57:46 +02:00
7 changed files with 193 additions and 26 deletions

View File

@@ -41,6 +41,7 @@ func Index(conf *cfg.Config) *cli.Command {
IndexClose(conf), IndexClose(conf),
IndexFields(conf), IndexFields(conf),
IndexIlm(conf), IndexIlm(conf),
IndexDu(conf),
// sub commands // sub commands
IndexAlias(conf), IndexAlias(conf),
@@ -277,3 +278,24 @@ func IndexIlm(conf *cfg.Config) *cli.Command {
}, },
} }
} }
func IndexDu(conf *cfg.Config) *cli.Command {
return &cli.Command{
Name: "du",
Usage: "show index disk usage",
UsageText: "index du <index>",
ShellComplete: func(ctx context.Context, cmd *cli.Command) {
complete(conf, cmd, Cindex)
},
Action: func(ctx context.Context, cmd *cli.Command) error {
index := cmd.Args().Get(0)
if index == "" {
return errors.New("no index specified")
}
return es.IndexDiskusage(conf, index)
},
}
}

View File

@@ -18,6 +18,8 @@ package es
import ( import (
"context" "context"
"encoding/json"
"errors"
"fmt" "fmt"
"log/slog" "log/slog"
"regexp" "regexp"
@@ -28,6 +30,7 @@ import (
"codeberg.org/scip/esctl/pkg/cfg" "codeberg.org/scip/esctl/pkg/cfg"
"codeberg.org/scip/esctl/pkg/printer" "codeberg.org/scip/esctl/pkg/printer"
"github.com/charmbracelet/lipgloss"
"github.com/elastic/go-elasticsearch/v9/typedapi/cat/indices" "github.com/elastic/go-elasticsearch/v9/typedapi/cat/indices"
"github.com/elastic/go-elasticsearch/v9/typedapi/esdsl" "github.com/elastic/go-elasticsearch/v9/typedapi/esdsl"
"github.com/elastic/go-elasticsearch/v9/typedapi/types/enums/healthstatus" "github.com/elastic/go-elasticsearch/v9/typedapi/types/enums/healthstatus"
@@ -306,3 +309,79 @@ func IndexFields(conf *cfg.Config, index string) error {
return nil return nil
} }
type Diskusage struct {
Total int64 `json:"total_in_bytes"`
Points int64 `json:"points_in_bytes"`
Norms int64 `json:"norms_in_bytes"`
TermVectors int64 `json:"term_vectors_in_bytes"`
KnnVectors int64 `json:"knn_vectors_in_bytes"`
BloomFilter int64 `json:"bloom_filter_in_bytes"`
}
type IndexDiskUsage struct {
AllFields Diskusage `json:"all_fields"`
Fields map[string]Diskusage `json:"fields"`
}
type ResIndexDiskUsage map[string]IndexDiskUsage
func IndexDiskusage(conf *cfg.Config, index string) error {
var bold = lipgloss.NewStyle().Bold(true)
res, err := conf.DefaultCluster.ES().Indices.DiskUsage(index).
RunExpensiveTasks(true).
Do(context.Background())
if err != nil {
return fmt.Errorf("failed to retrieve index disk usage: %w", esErrorString(err))
}
duRes := ResIndexDiskUsage{}
if err := json.Unmarshal(res, &duRes); err != nil {
return fmt.Errorf("failed to unmarshal disk usage response: %w", err)
}
diskusage, exists := duRes[index]
if !exists {
return errors.New("no disk usage reported for index")
}
table := printer.NewTableEmpty(conf).
WithHeaders("field", "bloom filter", "norms", "points", "term vectors", "knn vectors", "total")
for name, field := range diskusage.Fields {
if strings.HasPrefix(name, "_") || strings.HasSuffix(name, ".keyword") {
continue
}
table.AddRow(
name,
printer.Bytes(field.BloomFilter),
printer.Bytes(field.Norms),
printer.Bytes(field.Points),
printer.Bytes(field.TermVectors),
printer.Bytes(field.KnnVectors),
printer.Bytes(field.Total),
)
}
all := diskusage.AllFields
table.Sort()
table.AddRowLate(
bold.Render("Summary"),
printer.Bytes(all.BloomFilter),
printer.Bytes(all.Norms),
printer.Bytes(all.Points),
printer.Bytes(all.TermVectors),
printer.Bytes(all.KnnVectors),
printer.Bytes(all.Total),
)
if err := table.Print(); err != nil {
return err
}
return nil
}

View File

@@ -26,7 +26,6 @@ import (
"codeberg.org/scip/esctl/pkg/cfg" "codeberg.org/scip/esctl/pkg/cfg"
"codeberg.org/scip/esctl/pkg/printer" "codeberg.org/scip/esctl/pkg/printer"
"github.com/dustin/go-humanize"
) )
func NodeList(conf *cfg.Config) error { func NodeList(conf *cfg.Config) error {
@@ -114,18 +113,27 @@ func NodeShow(conf *cfg.Config, nodename string) error {
} }
k8snode := info.Attributes["k8s_node_name"] k8snode := info.Attributes["k8s_node_name"]
rank := "none"
adsel, exists := stat.AdaptiveSelection[id]
if exists {
rank = *adsel.Rank
}
table.Entries = [][]any{ table.Entries = [][]any{
{"Id", id}, {"Id", id},
{"Name", nodename}, {"Name", nodename},
{"Kubernetes node", k8snode}, {"Kubernetes node", k8snode},
{"Ip address", info.Ip}, {"Ip address", info.Ip},
{"Node rank", *stat.AdaptiveSelection[id].Rank}, {"Node rank", rank},
{"JVM", info.Jvm.VmName + " " + info.Jvm.Version}, {"JVM", info.Jvm.VmName + " " + info.Jvm.Version},
{"JVM Started", time.UnixMilli(info.Jvm.StartTimeInMillis)}, {"JVM Started", time.UnixMilli(info.Jvm.StartTimeInMillis)},
{"OS", info.Os.PrettyName + " " + info.Os.Version}, {"OS", info.Os.PrettyName + " " + info.Os.Version},
{"Node roles", roles}, {"Node roles", roles},
{"Node version", info.Version}, {"Node version", info.Version},
// FIXME: not implemented upstream
// see: https://github.com/elastic/go-elasticsearch/issues/1526
// {"Allocated shards", stat.Allocations.XXX},
{"HTTP clients", *stat.Http.CurrentOpen}, {"HTTP clients", *stat.Http.CurrentOpen},
{"CPUs", *info.Os.AllocatedProcessors}, {"CPUs", *info.Os.AllocatedProcessors},
{"Load 15m/5m/1m", fmt.Sprintf("%.2f/%.2f/%.2f", {"Load 15m/5m/1m", fmt.Sprintf("%.2f/%.2f/%.2f",
@@ -134,18 +142,41 @@ func NodeShow(conf *cfg.Config, nodename string) error {
stat.Os.Cpu.LoadAverage["1m"], stat.Os.Cpu.LoadAverage["1m"],
)}, )},
{"Open FD's", *stat.Process.OpenFileDescriptors}, {"Open FD's", *stat.Process.OpenFileDescriptors},
{"Response time avg", fmt.Sprintf("%dns", *stat.AdaptiveSelection[id].AvgResponseTimeNs)}, {"HTTP sesssions current/total", fmt.Sprintf("%d/%d",
*stat.Http.CurrentOpen,
*stat.Http.TotalOpened,
)},
{"Traffic rx/tx",
printer.ByteString(*stat.Transport.RxSizeInBytes) + " / " + printer.ByteString(*stat.Transport.TxSizeInBytes)},
{"Response time avg", time.Duration(*stat.AdaptiveSelection[id].AvgResponseTimeNs)},
{"Memory usage (used/avail)", {"Memory usage (used/avail)",
humanize.Bytes(uint64(*stat.Os.Mem.UsedInBytes)) + " / " + humanize.Bytes(uint64(*stat.Os.Mem.TotalInBytes))}, printer.ByteString(*stat.Os.Mem.UsedInBytes) + " / " + printer.ByteString(*stat.Os.Mem.TotalInBytes)},
{"Search queries current/total", fmt.Sprintf("%d/%d",
stat.Indices.Search.QueryCurrent,
stat.Indices.Search.QueryTotal,
)},
{"Search efficiency", stat.Indices.Search.QueryTimeInMillis / stat.Indices.Search.QueryTotal},
{"Docs count", stat.Indices.Docs.Count},
{"Merges current/total", fmt.Sprintf("%d/%d",
stat.Indices.Merges.Current,
stat.Indices.Merges.Total,
)},
{"Merge docs count current/total", fmt.Sprintf("%d/%d",
stat.Indices.Merges.CurrentDocs,
stat.Indices.Merges.TotalDocs,
)},
{"Merge size current/total", fmt.Sprintf("%s/%s",
printer.ByteString(stat.Indices.Merges.CurrentSizeInBytes),
printer.ByteString(stat.Indices.Merges.TotalSizeInBytes),
)},
{"CircuitBreaker trip count", *stat.Breakers["fielddata"].Tripped},
} }
if len(stat.Fs.Data) > 0 { if len(stat.Fs.Data) > 0 {
fs := stat.Fs.Data[0] fs := stat.Fs.Data[0]
table.Entries = append(table.Entries, [][]any{ table.AddRow("Storage usage (used/avail)",
{"Storage usage (used/avail)", printer.ByteString(*fs.AvailableInBytes)+" / "+printer.ByteString(*fs.TotalInBytes))
humanize.Bytes(uint64(*fs.AvailableInBytes)) + " / " + humanize.Bytes(uint64(*fs.TotalInBytes))}, table.AddRow("Storage mount", *fs.Mount)
{"Storage mount", *fs.Mount},
}...)
} }
if err := table.Print(); err != nil { if err := table.Print(); err != nil {

View File

@@ -163,28 +163,35 @@ func ShardAllocation(conf *cfg.Config, index string) error {
slog.Debug("ES result", "explain", res) slog.Debug("ES result", "explain", res)
currentNode := res.CurrentNode
table := printer.NewTable(conf, 2, 10) table := printer.NewTable(conf, 2, 10)
table.Addheaders("shard allocation setting", "value") table.Addheaders("shard allocation setting", "value")
roles := make([]string, len(currentNode.Roles))
for idx, role := range currentNode.Roles {
roles[idx] = role.Name
}
table.Entries = [][]any{ table.Entries = [][]any{
{"Index", index}, {"Index", index},
{"Current state", res.CurrentState}, {"Current state", res.CurrentState},
{"Can rebalance cluster", res.CanRebalanceCluster.Name},
{"Can rebalance to another node", res.CanRebalanceToOtherNode.Name},
{"Can remain on current node", res.CanRemainOnCurrentNode.Name},
}
if res.CurrentNode != nil {
currentNode := res.CurrentNode
roles := make([]string, len(currentNode.Roles))
for idx, role := range currentNode.Roles {
roles[idx] = role.Name
}
table.Entries = append(table.Entries, [][]any{
{"Current node", currentNode.Name}, {"Current node", currentNode.Name},
{"Current k8s node", currentNode.Attributes["k8s_node_name"]}, {"Current k8s node", currentNode.Attributes["k8s_node_name"]},
{"Current node address", currentNode.TransportAddress}, {"Current node address", currentNode.TransportAddress},
{"Current node id", currentNode.Id}, {"Current node id", currentNode.Id},
{"Current node weight", currentNode.WeightRanking}, {"Current node weight", currentNode.WeightRanking},
{"Current node roles", roles}, {"Current node roles", roles},
{"Can rebalance cluster", res.CanRebalanceCluster.Name}, }...)
{"Can rebalance to another node", res.CanRebalanceToOtherNode.Name}, } else {
{"Can remain on current node", res.CanRemainOnCurrentNode.Name}, table.AddRow("Current node", "not currently assigned to any node")
} }
if res.CurrentState == "unassigned" { if res.CurrentState == "unassigned" {

View File

@@ -22,10 +22,16 @@ type ByteSize struct {
size uint64 size uint64
} }
func Bytes(size int64) ByteSize {
return ByteSize{size: uint64(size)}
}
func (b *ByteSize) String() string { func (b *ByteSize) String() string {
return humanize.Bytes(b.size) return humanize.Bytes(b.size)
} }
func Bytes(size int64) *ByteSize {
return &ByteSize{size: uint64(size)}
}
func ByteString(size int64) string {
b := ByteSize{size: uint64(size)}
return b.String()
}

View File

@@ -38,12 +38,18 @@ func any2string(in any) string {
return strconv.Itoa(val) return strconv.Itoa(val)
case float64: case float64:
return fmt.Sprintf("%.2f", val) return fmt.Sprintf("%.2f", val)
case float32:
return fmt.Sprintf("%.2f", val)
case []string: case []string:
return strings.Join(val, ",") return strings.Join(val, ",")
case ByteSize: case ByteSize:
return val.String() return val.String()
case *ByteSize:
return val.String()
case time.Time: case time.Time:
return val.Format("2006-01-02 15:04:05") return val.Format("2006-01-02 15:04:05")
case time.Duration:
return val.String()
case []byte: case []byte:
return string(val) return string(val)
case nil: case nil:

View File

@@ -206,6 +206,22 @@ func (table *Table) AddRow(fields ...any) {
table.Entries = append(table.Entries, fields) table.Entries = append(table.Entries, fields)
} }
func (table *Table) AddRowLate(fields ...any) {
table.AddRow(fields)
if !table.processed {
return
}
row := make([]string, len(fields))
for idx, field := range fields {
row[idx] = any2string(field)
}
table.rows = append(table.rows, row)
}
// needed for json and yaml output // needed for json and yaml output
func (table *Table) toMap() []map[string]any { func (table *Table) toMap() []map[string]any {
raw := make([]map[string]any, len(table.Entries)) raw := make([]map[string]any, len(table.Entries))