From b0b01941b80e9b01f5db28a44ba8989aae16ae0a Mon Sep 17 00:00:00 2001 From: Thomas von Dein Date: Thu, 16 Jul 2026 23:15:13 +0200 Subject: [PATCH] add index copy (aka reindex) --- pkg/es/index_copy.go | 83 ++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 83 insertions(+) create mode 100644 pkg/es/index_copy.go diff --git a/pkg/es/index_copy.go b/pkg/es/index_copy.go new file mode 100644 index 0000000..63c5db6 --- /dev/null +++ b/pkg/es/index_copy.go @@ -0,0 +1,83 @@ +/* +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 . +*/ +package es + +import ( + "context" + "fmt" + "time" + + "codeberg.org/scip/esctl/pkg/cfg" + "codeberg.org/scip/esctl/pkg/printer" + "github.com/elastic/go-elasticsearch/v9/typedapi/esdsl" + "github.com/elastic/go-elasticsearch/v9/typedapi/types/enums/conflicts" +) + +func IndexCopy(conf *cfg.Config) error { + copy := conf.DefaultCluster.ES().Reindex() + + if conf.Force { + copy.Conflicts(conflicts.Proceed) + } + + if conf.RequestsPerSecond != 0 { + copy.RequestsPerSecond(fmt.Sprintf("%.2f", conf.RequestsPerSecond)) + } + + if conf.MaxDocs > 0 { + copy.MaxDocs(conf.MaxDocs) + } + + if conf.Timeout > 0 { + copy.Timeout(formatDuration(conf.Timeout)) + } + + if conf.Wait { + copy.WaitForActiveShards("all") + } + + if conf.Refresh { + copy.Refresh(true) + } + + copy.Source(esdsl.NewReindexSource().Index(conf.SourceIndices...)) + copy.Dest(esdsl.NewReindexDestination().Index(conf.Index)) + + res, err := copy.Do(context.Background()) + if err != nil { + return fmt.Errorf("failed to copy indices: %w", esErrorString(err)) + } + + table := printer.NewTable(conf, 2, 0). + WithHeaders("Reindex metrtic", "value") + + table.Entries = [][]any{ + {"Source indices", conf.SourceIndices}, + {"Target index", conf.Index}, + {"Batches", *res.Batches}, + {"Documents total", *res.Total}, + {"Documents created", *res.Created}, + {"Documents deleted", *res.Deleted}, + {"Documents updated", *res.Updated}, + {"Requests/s", *res.RequestsPerSecond}, + {"Timed out", *res.TimedOut}, + {"Time elapsed", time.Duration(*res.Took) * time.Millisecond}, + {"Version conflicts", *res.VersionConflicts}, + } + + return table.Print() +}