117 lines
2.6 KiB
Go
117 lines
2.6 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"flag"
|
|
"fmt"
|
|
"os/signal"
|
|
"syscall"
|
|
"time"
|
|
|
|
elasticsearch8 "github.com/elastic/go-elasticsearch/v8"
|
|
"github.com/elastic/go-elasticsearch/v8/typedapi/core/search"
|
|
"github.com/elastic/go-elasticsearch/v8/typedapi/types"
|
|
)
|
|
|
|
func main() {
|
|
var hostname string
|
|
var list, ha bool
|
|
flag.StringVar(&hostname, "host", "elastic.zerops", "hostname to connect to elastic")
|
|
flag.BoolVar(&list, "list", false, "whether to list the whole index")
|
|
flag.BoolVar(&ha, "ha", false, "is the cluster ha?")
|
|
flag.Parse()
|
|
|
|
ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
|
|
defer stop()
|
|
|
|
client, err := elasticsearch8.NewTypedClient(elasticsearch8.Config{
|
|
Addresses: []string{fmt.Sprintf("http://%s:9200", hostname)},
|
|
// Logger: &elastictransport.TextLogger{Output: os.Stdout, EnableRequestBody: true},
|
|
})
|
|
if err != nil {
|
|
panic(err)
|
|
}
|
|
|
|
const indexName = "my_index"
|
|
indicesExistsResponse, err := client.Indices.Exists(indexName).Do(ctx)
|
|
if err != nil {
|
|
panic(err)
|
|
}
|
|
if !indicesExistsResponse {
|
|
indicesCreateResponse, err := client.Indices.Create(indexName).Do(ctx)
|
|
if err != nil {
|
|
panic(err)
|
|
}
|
|
fmt.Println("created:", indicesCreateResponse.Index)
|
|
if !ha {
|
|
_, err = client.Indices.PutSettings().Indices(indexName).NumberOfReplicas("0").Do(ctx)
|
|
if err != nil {
|
|
panic(err)
|
|
}
|
|
}
|
|
} else {
|
|
fmt.Println("already exists:", indexName)
|
|
}
|
|
|
|
if list {
|
|
size := 10_000
|
|
searchResponse, err := client.Search().
|
|
Index(indexName).
|
|
Request(
|
|
&search.Request{
|
|
Query: &types.Query{
|
|
MatchAll: types.NewMatchAllQuery(),
|
|
},
|
|
Size: &size,
|
|
},
|
|
).
|
|
Do(ctx)
|
|
if err != nil {
|
|
panic(err)
|
|
}
|
|
for _, hit := range searchResponse.Hits.Hits {
|
|
var data map[string]any
|
|
_ = json.Unmarshal(hit.Source_, &data)
|
|
fmt.Printf("==> %s\n", data["message"])
|
|
}
|
|
}
|
|
|
|
for {
|
|
if ctx.Err() != nil {
|
|
fmt.Println("context canceled")
|
|
return
|
|
}
|
|
|
|
message := time.Now().Format(time.TimeOnly)
|
|
document := map[string]any{
|
|
"name": "yahoo",
|
|
"message": message,
|
|
}
|
|
indexResponse, err := client.Index(indexName).Request(document).Do(ctx)
|
|
if err != nil {
|
|
panic(err)
|
|
}
|
|
fmt.Printf("%s ==>\n", message)
|
|
|
|
getResponse, err := client.Get(indexName, indexResponse.Id_).Do(ctx)
|
|
if err != nil {
|
|
panic(err)
|
|
}
|
|
if !getResponse.Found {
|
|
fmt.Println("error: document not found")
|
|
continue
|
|
}
|
|
if getResponse.Id_ != indexResponse.Id_ {
|
|
fmt.Println("error: different ids")
|
|
continue
|
|
}
|
|
|
|
var data map[string]any
|
|
_ = json.Unmarshal(getResponse.Source_, &data)
|
|
fmt.Printf("==> %s\n", data["message"])
|
|
|
|
time.Sleep(time.Second)
|
|
}
|
|
}
|