wip: alpha version

internal/controller/input/es7/client.go

@@ -0,0 +1,108 @@
+package es7
+
+import (
+	"context"
+	"crypto/tls"
+	"fmt"
+	elastic "github.com/elastic/go-elasticsearch/v7"
+	"github.com/elastic/go-elasticsearch/v7/esapi"
+	"github.com/sirupsen/logrus"
+	"net/http"
+)
+
+type Client struct {
+	Endpoints []string `json:"endpoints"`
+	Username  string   `json:"username"`
+	Password  string   `json:"password"`
+	CA        string   `json:"ca"`
+
+	cli *elastic.Client
+}
+
+func (c *Client) InitClient(ctx context.Context) error {
+	var (
+		errCh    = make(chan error)
+		cliCh    = make(chan *elastic.Client)
+		hiddenCa = func(cs string) string {
+			if len(cs) > 0 {
+				return "******"
+			}
+
+			return "nil"
+		}
+	)
+
+	logrus.Debugf("es7.NewClient: endpoints=%v (username=%s password=%s ca=%s)", c.Endpoints, c.Username, c.Password, hiddenCa(c.CA))
+
+	ncFunc := func(endpoints []string, username, password, ca string) {
+		var (
+			err      error
+			cli      *elastic.Client
+			infoResp *esapi.Response
+		)
+
+		if cli, err = elastic.NewClient(
+			elastic.Config{
+				Addresses:     endpoints,
+				Username:      username,
+				Password:      password,
+				CACert:        []byte(c.CA),
+				RetryOnStatus: []int{429},
+				MaxRetries:    3,
+				RetryBackoff:  nil,
+				Transport: &http.Transport{
+					TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
+				},
+			},
+		); err != nil {
+			logrus.Debugf("es7.NewClient: elastic new client with endponts=%v err=%v", endpoints, err)
+			errCh <- err
+			return
+		}
+
+		if infoResp, err = cli.Info(); err != nil {
+			logrus.Debugf("es7.NewClient: ping err=%v", err)
+			errCh <- err
+			return
+		}
+
+		if infoResp.StatusCode != 200 {
+			err = fmt.Errorf("info es status=%d", infoResp.StatusCode)
+			logrus.Debugf("es7.NewClient: status err=%v", err)
+			errCh <- err
+			return
+		}
+
+		cliCh <- cli
+	}
+
+	go ncFunc(c.Endpoints, c.Username, c.Password, c.CA)
+
+	select {
+	case <-ctx.Done():
+		return fmt.Errorf("dial es=%s err=%v", c.Endpoints, context.DeadlineExceeded)
+	case c.cli = <-cliCh:
+		return nil
+	case e := <-errCh:
+		return e
+	}
+}
+
+func (c *Client) Ping(ctx context.Context) error {
+	rr, err := c.cli.Info(
+		c.cli.Info.WithContext(ctx),
+	)
+	if err != nil {
+		return err
+	}
+
+	if rr.StatusCode != 200 {
+		return fmt.Errorf("ping status=%d msg=%s", rr.StatusCode, rr.String())
+	}
+
+	return nil
+}
+
+func (c *Client) Save(ctx context.Context) error {
+	return nil
+}