Skip to content
Snippets Groups Projects
Select Git revision
  • 4a4d4a2b9cbefee413a0ce1acaa5d699200d4636
  • main default protected
  • v1.298.0
  • v1.297.0
  • v1.296.0
  • v1.295.0
  • v1.294.0
  • v1.293.0
  • v1.292.0
  • v1.291.0
  • v1.290.0
  • v1.289.0
  • v1.288.0
  • v1.287.0
  • v1.286.0
  • v1.285.0
  • v1.284.0
  • v1.283.0
  • v1.282.0
  • v1.281.0
  • v1.280.0
  • v1.279.0
22 results

writer.go

Blame
  • writer.go 4.07 KiB
    package search
    
    import (
    	"crypto/tls"
    	"encoding/json"
    	opensearch "github.com/opensearch-project/opensearch-go"
    	opensearchapi "github.com/opensearch-project/opensearch-go/opensearchapi"
    	"gitlab.bob.co.za/bob-public-utils/bobgroup-go-utils/errors"
    	"gitlab.bob.co.za/bob-public-utils/bobgroup-go-utils/logs"
    	"net/http"
    	"strings"
    	"time"
    )
    
    type Writer struct {
    	config              Config
    	client              *opensearch.Client
    	api                 *opensearchapi.API
    	timeSeriesByName    map[string]TimeSeries
    	documentStoreByName map[string]DocumentStore
    }
    
    func New(config Config) (Writer, error) {
    	w := Writer{}
    
    	if err := config.Validate(); err != nil {
    		return w, errors.Wrapf(err, "invalid config")
    	}
    	w = Writer{
    		config:              config,
    		timeSeriesByName:    map[string]TimeSeries{},
    		documentStoreByName: map[string]DocumentStore{},
    	}
    
    	searchConfig := opensearch.Config{
    		Transport: &http.Transport{
    			TLSClientConfig:       &tls.Config{InsecureSkipVerify: true},
    			ResponseHeaderTimeout: time.Second * 28, // timeout the request just before api gateway times out
    		},
    		Addresses: config.Addresses,
    		Username:  config.Username,
    		Password:  config.Password,
    	}
    	// Initialize the client with SSL/TLS enabled.
    	var err error
    	w.client, err = opensearch.NewClient(searchConfig)
    	if err != nil {
    		return w, errors.Wrapf(err, "cannot initialize opensearch connection")
    	}
    	// Print OpenSearch version information on console.
    	logs.Info("Search client created with config: %+v", searchConfig)
    
    	w.api = opensearchapi.New(w.client)
    
    	return w, nil
    }
    
    func (writer Writer) Write(indexName string, id string, doc interface{}) (*IndexResponse, error) {
    	if writer.client == nil {
    		return nil, errors.Errorf("writer closed")
    	}
    	jsonDocStr, ok := doc.(string)
    	if !ok {
    		jsonDoc, err := json.Marshal(doc)
    		if err != nil {
    			return nil, errors.Wrapf(err, "failed to JSON encode document")
    		}
    		jsonDocStr = string(jsonDoc)
    	}
    	options := []func(*opensearchapi.IndexRequest){}