elasticsearch put index

elasticput.go

package main

import (
	"encoding/csv"
	"encoding/json"
	"flag"
	"fmt"
	"io"
	"io/ioutil"
	"log"
	"net/http"
	"os"
	"strconv"
	"strings"
	"sync"
	"time"
)

var (
	uri   string
	index string

	client = http.Client{Timeout: time.Duration(time.Second)}
	wg     = &sync.WaitGroup{}
)

func csv2json(record []string) string {
	row := make(map[string]string)
	for i, val := range record {
		row["col_"+strconv.Itoa(i)] = val
	}
	b, err := json.Marshal(row)
	if err != nil {
		log.Fatal(err)
	}
	return string(b)
}

func put(id int, payload string) {
	req, err := http.NewRequest("PUT", uri+"/"+index+"/document/"+strconv.Itoa(id), strings.NewReader(payload))
	if err != nil {
		log.Fatal(err)
	}

	resp, err := client.Do(req)
	if err != nil {
		log.Fatal(err)
	}

	if body, err := ioutil.ReadAll(resp.Body); err != nil {
		log.Fatal(err)
	} else {
		log.Println(string(body))
	}
	resp.Body.Close()
	wg.Done()
}

func main() {
	flag.StringVar(&uri, "uri", "http://localhost:9200", "elastic search URI")
	flag.StringVar(&index, "index", "csv", "index name")

	flag.Usage = func() {
		fmt.Fprintf(os.Stderr, "usage: %s [flags]\n", os.Args[0])
		fmt.Fprintln(os.Stderr, "flags:")
		flag.PrintDefaults()
		os.Exit(2)
	}
	flag.Parse()

	r := csv.NewReader(os.Stdin)
	for id := 1; ; id++ {
		record, err := r.Read()
		if err == io.EOF {
			break
		}
		if err != nil {
			log.Fatal(err)
		}

		wg.Add(1)
		payload := csv2json(record)
		go put(id, payload)
	}

	wg.Wait()
}