Sample code

issue-892.go

// Copyright 2012-present Oliver Eilhard. All rights reserved.
// Use of this source code is governed by a MIT-license.
// See http://olivere.mit-license.org/license.txt for details.

// Connect simply connects to Elasticsearch.
//
// Example
//
//
//     connect -url=http://127.0.0.1:9200 -sniff=false
//
package main

import (
	"context"
	"flag"
	"fmt"
	"log"
	"math/rand"
	"os"
	"time"

	"github.com/olivere/elastic"
)

const (
	mapping = `
	{
		"settings":{
			"number_of_shards":1,
			"number_of_replicas":0
		},
		"mappings":{
			"_doc":{
				"properties":{
					"status":{
						"type":"keyword"
					},
					"queued":{
						"type":"integer"
					},
					"slot":{
						"type":"keyword"
					}
				}
			}
		}
	}`
)

var (
	Statuses = []string{"succeeded", "delayed", "failed", "working"}
)

type Doc struct {
	Status    string `json:"status"`
	Timestamp int64  `json:"@timestamp"`
	Slot      string `json:"slot,omitempty"`
}

func main() {
	var (
		url   = flag.String("url", "http://localhost:9200", "Elasticsearch URL")
		sniff = flag.Bool("sniff", true, "Enable or disable sniffing")
		index = flag.String("index", "", "Index name")
		debug = flag.Bool("d", false, "Enable debug output")
	)
	flag.Parse()
	log.SetFlags(0)

	if *url == "" {
		*url = "http://127.0.0.1:9200"
	}
	if *index == "" {
		log.Fatal("please specify an index name -index")
	}

	// Create an Elasticsearch client
	options := []elastic.ClientOptionFunc{
		elastic.SetURL(*url),
		elastic.SetSniff(*sniff),
	}
	if *debug {
		logger := log.New(os.Stdout, "", 0)
		options = append(options, elastic.SetTraceLog(logger))
	}
	client, err := elastic.NewClient(options...)
	if err != nil {
		log.Fatal(err)
	}

	// Just a status message
	fmt.Println("Connection succeeded")

	// Check if index already exists. We'll drop it then.
	// Next, we create a fresh index/mapping.
	exists, err := client.IndexExists(*index).Do(context.Background())
	if err != nil {
		log.Fatal(err)
	}
	if exists {
		_, err := client.DeleteIndex(*index).Do(context.Background())
		if err != nil {
			log.Fatal(err)
		}
	}
	_, err = client.CreateIndex(*index).Body(mapping).Do(context.Background())
	if err != nil {
		log.Fatal(err)
	}

	// Add some docs via Bulk API
	{
		var reqs []elastic.BulkableRequest
		for i := 0; i < 1000; i++ {
			doc := Doc{
				Status:    Statuses[rand.Intn(len(Statuses))],
				Timestamp: time.Now().Add(time.Duration(-i)*time.Minute).UnixNano() / 1e9,
				Slot:      fmt.Sprintf("slot-%d", rand.Int63n(10)),
			}
			req := elastic.NewBulkIndexRequest().
				Id(fmt.Sprint(i)).
				Doc(doc)
			reqs = append(reqs, req)
		}
		// Write the bulk requests and make sure everything's written
		_, err := client.Bulk().
			Index(*index).
			Type("_doc").
			Add(reqs...).
			Refresh("wait_for").
			Do(context.Background())
		if err != nil {
			log.Fatal(err)
		}
	}

	// Now run a query
	today := time.Now().UTC().Truncate(24 * time.Hour)
	yesterday := today.AddDate(0, 0, -1)
	matchQ := elastic.NewQueryStringQuery("status:succeeded OR status:delayed")
	rangeQ := elastic.NewRangeQuery("@timestamp").
		Gte(yesterday.UnixNano() / 1e9).
		Lte(today.UnixNano() / 1e9)
	query := elastic.NewBoolQuery().
		Must(matchQ).
		Filter(rangeQ)
	res, err := client.Search(*index).
		Type("_doc").
		Query(query).
		Pretty(true).
		Do(context.Background())
	if err != nil {
		log.Fatal(err)
	}
	if n := res.TotalHits(); n > 0 {
		log.Printf("Found a total of %d records\n", n)
	}
}