from domain to serverless

migrate.go

func CopyIndex(index CatIndex, domain *opensearch.Client, serverless *opensearch.Client) (CopyResult, error) {

	results := CopyResult{
		Index:    index.Index,
		Docs:     0,
		Duration: time.Duration(0),
	}

	start := time.Now()

	indexExists, err := serverless.Indices.Exists([]string{"initial-access"})

	if err != nil {
		return results, err
	}

	if indexExists.StatusCode != 200 {
		mappings, err := GetMappings(domain, index.Index)
		if err != nil {
			return results, err
		}
		if err := CreateIndex(serverless, index.Index, mappings); err != nil {
			return results, err
		}
	}

	scrollId := ""

	indexer, err := opensearchutil.NewBulkIndexer(opensearchutil.BulkIndexerConfig{
		Client:     serverless,
		Index:      index.Index,
		NumWorkers: 96,
	})

	if err != nil {
		return results, err
	}

	for {
		result, err := Search(domain, index.Index, scrollId)
		if err != nil {
			return results, err
		}
		scrollId = result.ScrollID
		if len(result.Hits.Hits) == 0 {
			break
		}

		for _, hit := range result.Hits.Hits {
			if err := indexer.Add(
				context.Background(),
				opensearchutil.BulkIndexerItem{
					Action:     "index",
					Body:       strings.NewReader(string(hit.Source)),
					DocumentID: hit.ID,
				},
			); err != nil {
				return results, err
			}
			results.Docs++
		}
	}

	if err := indexer.Close(context.Background()); err != nil {
		return results, err
	}

	stats := indexer.Stats()
	after := time.Since(start)
	log.Info().
		Str("index", index.Index).
		Bool("exists", indexExists.StatusCode == 200).
		Str("duration", after.String()).
		Uint64("failed", stats.NumFailed).
		Uint64("indexed", stats.NumFlushed).
		Msg("indexed documents")

	results.Duration = after

	return results, nil
}