// SendDocument sends a batch of documents to Rockset
SendDocument sends a batch of documents to Rockset
// SendDocument sends a batch of documents to Rockset
func (e *Elastic) SendDocument(docs []interface{}) error {
	numDocs := len(docs)
	numEventIngested.Add(float64(numDocs))
	var builder bytes.Buffer
	for i := 0; i < len(docs); i++ {
		line, err := json.Marshal(docs[i])
		if err != nil {
			return fmt.Errorf("failed to marshal document: %w", err)
		}
		index := make(map[string]interface{})
		index["_index"] = e.esIndexName
		index["_id"] = guuid.New().String()
		ret := make(map[string]interface{})
		ret["index"] = index
		metaLine, err := json.Marshal(ret)
		if err != nil {
			return fmt.Errorf("failed to marshal request: %w", err)
		}
		builder.Write(metaLine)
		builder.WriteByte('\n')
		builder.Write(line)
		builder.WriteByte('\n')
	}
	body := builder.Bytes()
	bulkURL := e.esURL + "/_bulk"
	elasticHTTPRequest, _ := http.NewRequest(http.MethodPost, bulkURL, bytes.NewBuffer(body))
	elasticHTTPRequest.Header.Add("Authorization", e.esAuth)
	elasticHTTPRequest.Header.Add("Content-Type", "application/x-ndjson")
	resp, err := e.client.Do(elasticHTTPRequest)
	if err != nil {
		recordWritesErrored(float64(numDocs))
		return fmt.Errorf("failed to send request: %w", err)
	}
	defer deferredErrorCloser(resp.Body)
	if resp.StatusCode != http.StatusOK {
		recordWritesErrored(float64(numDocs))
		bodyBytes, err := ioutil.ReadAll(resp.Body)
		if err != nil {
			return fmt.Errorf("failed to read response body: %w", err)
		}
		return fmt.Errorf("error code: %d, body: %s", resp.StatusCode, string(bodyBytes))
	}
	recordWritesCompleted(float64(numDocs))
	return nil
}
func ( e * Elastic ) SendDocument ( docs [ ] interface { } ) error { numDocs := len ( docs ) numEventIngested . Add ( float64 ( numDocs ) ) var builder bytes . Buffer for i := 0 ; i < len ( docs ) ; i ++ { line , err := json . Marshal ( docs [ i ] ) if err != nil { return fmt . Errorf ( " " , err ) } index := make ( map [ string ] interface { } ) index [ " " ] = e . esIndexName index [ " " ] = guuid . New ( ) . String ( ) ret := make ( map [ string ] interface { } ) ret [ " " ] = index metaLine , err := json . Marshal ( ret ) if err != nil { return fmt . Errorf ( " " , err ) } builder . Write ( metaLine ) builder . WriteByte ( '\n' ) builder . Write ( line ) builder . WriteByte ( '\n' ) } body := builder . Bytes ( ) bulkURL := e . esURL + " " elasticHTTPRequest , _ := http . NewRequest ( http . MethodPost , bulkURL , bytes . NewBuffer ( body ) ) elasticHTTPRequest . Header . Add ( " " , e . esAuth ) elasticHTTPRequest . Header . Add ( " " , " " ) resp , err := e . client . Do ( elasticHTTPRequest ) if err != nil { recordWritesErrored ( float64 ( numDocs ) ) return fmt . Errorf ( " " , err ) } defer deferredErrorCloser ( resp . Body ) if resp . StatusCode != http . StatusOK { recordWritesErrored ( float64 ( numDocs ) ) bodyBytes , err := ioutil . ReadAll ( resp . Body ) if err != nil { return fmt . Errorf ( " " , err ) } return fmt . Errorf ( " " , resp . StatusCode , string ( bodyBytes ) ) } recordWritesCompleted ( float64 ( numDocs ) ) return nil }
SendDocument sends a batch of documents to Rockset
SendDocument sends a batch of documents to Rockset