elasticstream/cmd/insert.go

package main

import (
	"bytes"
	"context"
	"encoding/json"
	"flag"
	"fmt"
	"log"
	"time"

	"github.com/elastic/go-elasticsearch/esapi"
	"github.com/elastic/go-elasticsearch/v8"
)

func insert(client *elasticsearch.Client, index string, record []byte) error {
	req := esapi.IndexRequest{
		Index: index,
		// DocumentID: fmt.Sprintf("%d", u.ID),
		Body:    bytes.NewReader(record),
		Refresh: "true",
	}

	res, err := req.Do(context.Background(), client)
	if err != nil {
		return err
	}
	defer res.Body.Close()

	if res.IsError() {
		return fmt.Errorf("error indexing document: %s", res.String())
	}

	fmt.Printf("Document created: %s\n", res.String())
	return nil
}

type Record struct {
	ID        int       `json:"id"`
	Name      string    `json:"name"`
	Price     float64   `json:"price"`
	Timestamp time.Time `json:"timestamp"`
}

func main() {
	var count int
	flag.IntVar(&count, "count", 10, "number of records to generate")

	var index string
	flag.StringVar(&index, "index", "default", "index name")

	var host string
	flag.StringVar(&host, "host", "http://localhost:9200", "host")

	flag.Parse()

	fmt.Println("count:", count)
	fmt.Println("index:", index)
	fmt.Println("host:", host)

	cfg := elasticsearch.Config{
		Addresses: []string{
			host,
		},
	}

	client, err := elasticsearch.NewClient(cfg)
	if err != nil {
		log.Println("elasticsearch.NewDefaultClient() err:", err)
		return
	}

	for i := 0; i < count; i++ {
		record := Record{
			ID:        i,
			Price:     float64(i * 10),
			Name:      fmt.Sprintf("%s_%d", index, i),
			Timestamp: time.Now(),
		}

		bytes, _ := json.Marshal(record)

		err := insert(client, index, bytes)
		if err != nil {
			log.Println("insert() err:", err)
			continue
		}
	}
}