<commit_before>package elasticsearch

import (
	"context"
	"encoding/json"
	"fmt"
	"io"
	"log"

	opensearch "github.com/opensearch-project/opensearch-go"
	opensearchapi "github.com/opensearch-project/opensearch-go/opensearchapi"
	opensearchutil "github.com/opensearch-project/opensearch-go/opensearchutil"

	"go.opentelemetry.io/otel/api/trace"
	"go.opentelemetry.io/otel/codes"

	"github.com/ipfs-search/ipfs-search/components/index"
	"github.com/ipfs-search/ipfs-search/instr"
)

// Index wraps an Elasticsearch index to store documents
type Index struct {
	es  *opensearch.Client
	cfg *Config
	bi  opensearchutil.BulkIndexer

	*instr.Instrumentation
}

// New returns a new index.
func New(es *opensearch.Client, cfg *Config, i *instr.Instrumentation) index.Index {
	index := &Index{
		es:              es,
		cfg:             cfg,
		Instrumentation: i,
	}

	biCfg := index.getBulkIndexerCfg()
	bi, err := opensearchutil.NewBulkIndexer(biCfg)

	if err != nil {
		log.Fatalf("error getting bulk indexer: %v", err)
	}
	index.bi = bi

	return index
}

// String returns the name of the index, for convenient logging.
func (i *Index) String() string {
	return i.cfg.Name
}

type debugLogger struct{}

func (d debugLogger) Printf(s string, args ...interface{}) {
	log.Printf(s, args...)
}

func (i *Index) getBulkIndexerCfg() opensearchutil.BulkIndexerConfig {
	dl := debugLogger{}

	return opensearchutil.BulkIndexerConfig{
		Client:      i.es,
		DebugLogger: dl,
		OnError: func(ctx context.Context, err error) {
			span := trace.SpanFromContext(ctx)
			span.RecordError(ctx, err, trace.WithErrorStatus(codes.Error))
			log.Printf("error in ES bulk indexer: %v", err)
		},
		OnFlushStart: func(ctx context.Context) context.Context {
			newCtx, _ := i.Tracer.Start(ctx, "index.elasticsearch.BulkIndexerFlush")

			return newCtx
		},
		OnFlushEnd: func(ctx context.Context) {
			span := trace.SpanFromContext(ctx)
			// log.Printf("ES stats: %+v", )
			span.End()
		},
		Index: i.cfg.Name,
	}
}

// func successCb(
// 	ctx context.Context,
// 	item opensearchutil.BulkIndexerItem,
// 	res opensearchutil.BulkIndexerResponseItem,
// ) {
// 	fmt.Printf("[%d] %s test/%s", res.Status, res.Result, item.DocumentID)
// }

// func failureCb(
// 	ctx context.Context,
// 	item opensearchutil.BulkIndexerItem,
// 	res opensearchutil.BulkIndexerResponseItem, err error,
// ) {
// 	if err != nil {
// 		log.Printf("ERROR: %s", err)
// 	} else {
// 		log.Printf("ERROR: %s: %s", res.Error.Type, res.Error.Reason)
// 	}
// }

func (i *Index) bulkAction(
	ctx context.Context,
	action string,
	id string,
	properties interface{},
) error {
	var body io.Reader
	if properties != nil {
		body = opensearchutil.NewJSONReader(properties)
	}<commit_msg>Tweak verbosity for bulk indexer.<commit_after>package elasticsearch

import (
	"context"
	"encoding/json"
	"fmt"
	"io"
	"log"

	opensearch "github.com/opensearch-project/opensearch-go"
	opensearchapi "github.com/opensearch-project/opensearch-go/opensearchapi"
	opensearchutil "github.com/opensearch-project/opensearch-go/opensearchutil"

	"go.opentelemetry.io/otel/api/trace"
	"go.opentelemetry.io/otel/codes"

	"github.com/ipfs-search/ipfs-search/components/index"
	"github.com/ipfs-search/ipfs-search/instr"
)

// Index wraps an Elasticsearch index to store documents
type Index struct {
	es  *opensearch.Client
	cfg *Config
	bi  opensearchutil.BulkIndexer

	*instr.Instrumentation
}

// New returns a new index.
func New(es *opensearch.Client, cfg *Config, i *instr.Instrumentation) index.Index {
	index := &Index{
		es:              es,
		cfg:             cfg,
		Instrumentation: i,
	}

	biCfg := index.getBulkIndexerCfg()
	bi, err := opensearchutil.NewBulkIndexer(biCfg)

	if err != nil {
		log.Fatalf("error getting bulk indexer: %v", err)
	}
	index.bi = bi

	return index
}

// String returns the name of the index, for convenient logging.
func (i *Index) String() string {
	return i.cfg.Name
}

type debugLogger struct{}

// func (d debugLogger) Printf(s string, args ...interface{}) {
// 	log.Printf(s, args...)
// }

func (i *Index) getBulkIndexerCfg() opensearchutil.BulkIndexerConfig {
	return opensearchutil.BulkIndexerConfig{
		Client: i.es,
		// DebugLogger: debugLogger{},
		OnError: func(ctx context.Context, err error) {
			if err != nil {
				// Weirdly this gets called with nil error some times.
				span := trace.SpanFromContext(ctx)
				span.RecordError(ctx, err, trace.WithErrorStatus(codes.Error))
				log.Printf("Error flushing index buffer: %v", err)
			}
		},
		OnFlushStart: func(ctx context.Context) context.Context {
			newCtx, _ := i.Tracer.Start(ctx, "index.elasticsearch.BulkIndexerFlush")

			return newCtx
		},
		OnFlushEnd: func(ctx context.Context) {
			span := trace.SpanFromContext(ctx)
			log.Println("Flushed index buffer")

			// log.Printf("ES stats: %+v", )
			span.End()
		},
		Index: i.cfg.Name,
	}
}

// func successCb(
// 	ctx context.Context,
// 	item opensearchutil.BulkIndexerItem,
// 	res opensearchutil.BulkIndexerResponseItem,
// ) {
// 	fmt.Printf("[%d] %s test/%s", res.Status, res.Result, item.DocumentID)
// }

// func failureCb(
// 	ctx context.Context,
// 	item opensearchutil.BulkIndexerItem,
// 	res opensearchutil.BulkIndexerResponseItem, err error,
// ) {
// 	if err != nil {
// 		log.Printf("ERROR: %s", err)
// 	} else {
// 		// TODO: Catch 429 TOO MANY REQUESTS
// 		// Random exponential backoff sleep (blocks worker)
// 		log.Printf("ERROR: %s: %s", res.Error.Type, res.Error.Reason)
// 	}
// }

func (i *Index) bulkAction(
	ctx context.Context,
	action string,
	id string,
	properties interface{},
) error {
	var body io.Reader
	if properties != nil {
		body = opensearchutil.NewJSONReader(properties)
	}
{'fraction_non_alphanumeric': 0.14155571128241065, 'fraction_numerical': 0.0005255781359495446, 'mean_word_length': 3.6151980598221503, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5310869', 'n_tokens_mistral': 2182, 'n_tokens_neox': 2101, 'n_words': 450}