Go 本地調(diào)試Elasticsearch

   報(bào)錯(cuò)提示 :Error getting response: x509: certificate signed by unknown authority 加 InsecureSkipVerify: true 跳過TLS認(rèn)證.

Elasticsearch 8 以后默認(rèn)拒絕http請求連接 docker-compose 啟動(dòng)時(shí),若使用官方提供的docker-compose.yml 和 .env 則需要加 docker-compose up -d (不加 -d 會(huì)報(bào)健康檢查不通過);
注: 數(shù)據(jù)目錄建議掛載本地目錄,需要添加寫入權(quán)限
測試運(yùn)行環(huán)境 :CentOS Stream release 9

.env 內(nèi)容:

# Password for the 'elastic' user (at least 6 characters)
ELASTIC_PASSWORD=123456

# Password for the 'kibana_system' user (at least 6 characters)
KIBANA_PASSWORD=password

# Version of Elastic products
STACK_VERSION=8.5.2

# Set the cluster name
CLUSTER_NAME=docker-cluster

# Set to 'basic' or 'trial' to automatically start the 30-day trial
LICENSE=basic
#LICENSE=trial

# Port to expose Elasticsearch HTTP API to the host
ES_PORT=9200
#ES_PORT=127.0.0.1:9200

# Port to expose Kibana to the host
KIBANA_PORT=5601
#KIBANA_PORT=80

# Increase or decrease based on the available host memory (in bytes)
MEM_LIMIT=1073741824

# Project namespace (defaults to the current folder name if not set)
#COMPOSE_PROJECT_NAME=demo

docker-compose 文件內(nèi)容如下:

version: "2.2"

services:
  setup:
    image: docker.elastic.co/elasticsearch/elasticsearch:${STACK_VERSION}
    volumes:
      - ./certs:/usr/share/elasticsearch/config/certs
    user: "0"
    command: >
      bash -c '
        if [ x${ELASTIC_PASSWORD} == x ]; then
          echo "Set the ELASTIC_PASSWORD environment variable in the .env file";
          exit 1;
        elif [ x${KIBANA_PASSWORD} == x ]; then
          echo "Set the KIBANA_PASSWORD environment variable in the .env file";
          exit 1;
        fi;
        if [ ! -f config/certs/ca.zip ]; then
          echo "Creating CA";
          bin/elasticsearch-certutil ca --silent --pem -out config/certs/ca.zip;
          unzip config/certs/ca.zip -d config/certs;
        fi;
        if [ ! -f config/certs/certs.zip ]; then
          echo "Creating certs";
          echo -ne \
          "instances:\n"\
          "  - name: es01\n"\
          "    dns:\n"\
          "      - es01\n"\
          "      - localhost\n"\
          "    ip:\n"\
          "      - 127.0.0.1\n"\
          "  - name: es02\n"\
          "    dns:\n"\
          "      - es02\n"\
          "      - localhost\n"\
          "    ip:\n"\
          "      - 127.0.0.1\n"\
          "  - name: es03\n"\
          "    dns:\n"\
          "      - es03\n"\
          "      - localhost\n"\
          "    ip:\n"\
          "      - 127.0.0.1\n"\
          > config/certs/instances.yml;
          bin/elasticsearch-certutil cert --silent --pem -out config/certs/certs.zip --in config/certs/instances.yml --ca-cert config/certs/ca/ca.crt --ca-key config/certs/ca/ca.key;
          unzip config/certs/certs.zip -d config/certs;
        fi;
        echo "Setting file permissions"
        chown -R root:root config/certs;
        find . -type d -exec chmod 750 \{\} \;;
        find . -type f -exec chmod 640 \{\} \;;
        echo "Waiting for Elasticsearch availability";
        until curl -s --cacert config/certs/ca/ca.crt https://es01:9200 | grep -q "missing authentication credentials"; do sleep 30; done;
        echo "Setting kibana_system password";
        until curl -s -X POST --cacert config/certs/ca/ca.crt -u "elastic:${ELASTIC_PASSWORD}" -H "Content-Type: application/json" https://es01:9200/_security/user/kibana_system/_password -d "{\"password\":\"${KIBANA_PASSWORD}\"}" | grep -q "^{}"; do sleep 10; done;
        echo "All done!";
      '
    healthcheck:
      test: ["CMD-SHELL", "[ -f config/certs/es01/es01.crt ]"]
      interval: 1s
      timeout: 5s
      retries: 120

  es01:
    depends_on:
      setup:
        condition: service_healthy
    image: docker.elastic.co/elasticsearch/elasticsearch:${STACK_VERSION}
    volumes:
      - ./certs:/usr/share/elasticsearch/config/certs
      - ./esdata01:/usr/share/elasticsearch/data
    ports:
      - ${ES_PORT}:9200
    environment:
      - node.name=es01
      - cluster.name=${CLUSTER_NAME}
      - cluster.initial_master_nodes=es01,es02,es03
      - discovery.seed_hosts=es02,es03
      - ELASTIC_PASSWORD=${ELASTIC_PASSWORD}
      - bootstrap.memory_lock=true
      - xpack.security.enabled=true
      - xpack.security.http.ssl.enabled=true
      - xpack.security.http.ssl.key=certs/es01/es01.key
      - xpack.security.http.ssl.certificate=certs/es01/es01.crt
      - xpack.security.http.ssl.certificate_authorities=certs/ca/ca.crt
      - xpack.security.http.ssl.verification_mode=certificate
      - xpack.security.transport.ssl.enabled=true
      - xpack.security.transport.ssl.key=certs/es01/es01.key
      - xpack.security.transport.ssl.certificate=certs/es01/es01.crt
      - xpack.security.transport.ssl.certificate_authorities=certs/ca/ca.crt
      - xpack.security.transport.ssl.verification_mode=certificate
      - xpack.license.self_generated.type=${LICENSE}
    mem_limit: ${MEM_LIMIT}
    ulimits:
      memlock:
        soft: -1
        hard: -1
    healthcheck:
      test:
        [
          "CMD-SHELL",
          "curl -s --cacert config/certs/ca/ca.crt https://localhost:9200 | grep -q 'missing authentication credentials'",
        ]
      interval: 10s
      timeout: 10s
      retries: 120

  es02:
    depends_on:
      - es01
    image: docker.elastic.co/elasticsearch/elasticsearch:${STACK_VERSION}
    volumes:
      - ./certs:/usr/share/elasticsearch/config/certs
      - ./esdata02:/usr/share/elasticsearch/data
    environment:
      - node.name=es02
      - cluster.name=${CLUSTER_NAME}
      - cluster.initial_master_nodes=es01,es02,es03
      - discovery.seed_hosts=es01,es03
      - bootstrap.memory_lock=true
      - xpack.security.enabled=true
      - xpack.security.http.ssl.enabled=true
      - xpack.security.http.ssl.key=certs/es02/es02.key
      - xpack.security.http.ssl.certificate=certs/es02/es02.crt
      - xpack.security.http.ssl.certificate_authorities=certs/ca/ca.crt
      - xpack.security.http.ssl.verification_mode=certificate
      - xpack.security.transport.ssl.enabled=true
      - xpack.security.transport.ssl.key=certs/es02/es02.key
      - xpack.security.transport.ssl.certificate=certs/es02/es02.crt
      - xpack.security.transport.ssl.certificate_authorities=certs/ca/ca.crt
      - xpack.security.transport.ssl.verification_mode=certificate
      - xpack.license.self_generated.type=${LICENSE}
    mem_limit: ${MEM_LIMIT}
    ulimits:
      memlock:
        soft: -1
        hard: -1
    healthcheck:
      test:
        [
          "CMD-SHELL",
          "curl -s --cacert config/certs/ca/ca.crt https://localhost:9200 | grep -q 'missing authentication credentials'",
        ]
      interval: 10s
      timeout: 10s
      retries: 120

  es03:
    depends_on:
      - es02
    image: docker.elastic.co/elasticsearch/elasticsearch:${STACK_VERSION}
    volumes:
      - ./certs:/usr/share/elasticsearch/config/certs
      - ./esdata03:/usr/share/elasticsearch/data
    environment:
      - node.name=es03
      - cluster.name=${CLUSTER_NAME}
      - cluster.initial_master_nodes=es01,es02,es03
      - discovery.seed_hosts=es01,es02
      - bootstrap.memory_lock=true
      - xpack.security.enabled=true
      - xpack.security.http.ssl.enabled=true
      - xpack.security.http.ssl.key=certs/es03/es03.key
      - xpack.security.http.ssl.certificate=certs/es03/es03.crt
      - xpack.security.http.ssl.certificate_authorities=certs/ca/ca.crt
      - xpack.security.http.ssl.verification_mode=certificate
      - xpack.security.transport.ssl.enabled=true
      - xpack.security.transport.ssl.key=certs/es03/es03.key
      - xpack.security.transport.ssl.certificate=certs/es03/es03.crt
      - xpack.security.transport.ssl.certificate_authorities=certs/ca/ca.crt
      - xpack.security.transport.ssl.verification_mode=certificate
      - xpack.license.self_generated.type=${LICENSE}
    mem_limit: ${MEM_LIMIT}
    ulimits:
      memlock:
        soft: -1
        hard: -1
    healthcheck:
      test:
        [
          "CMD-SHELL",
          "curl -s --cacert config/certs/ca/ca.crt https://localhost:9200 | grep -q 'missing authentication credentials'",
        ]
      interval: 10s
      timeout: 10s
      retries: 120

  kibana:
    depends_on:
      es01:
        condition: service_healthy
      es02:
        condition: service_healthy
      es03:
        condition: service_healthy
    image: docker.elastic.co/kibana/kibana:${STACK_VERSION}
    volumes:
      - ./certs:/usr/share/kibana/config/certs
      - ./kibanadata:/usr/share/kibana/data
    ports:
      - ${KIBANA_PORT}:5601
    environment:
      - SERVERNAME=kibana
      - ELASTICSEARCH_HOSTS=https://es01:9200
      - ELASTICSEARCH_USERNAME=kibana_system
      - ELASTICSEARCH_PASSWORD=${KIBANA_PASSWORD}
      - ELASTICSEARCH_SSL_CERTIFICATEAUTHORITIES=config/certs/ca/ca.crt
    mem_limit: ${MEM_LIMIT}
    healthcheck:
      test:
        [
          "CMD-SHELL",
          "curl -s -I http://localhost:5601 | grep -q 'HTTP/1.1 302 Found'",
        ]
      interval: 10s
      timeout: 10s
      retries: 120

volumes:
  certs:
    driver: local
  esdata01:
    driver: local
  esdata02:
    driver: local
  esdata03:
    driver: local
  kibanadata:
    driver: local

main.go

package main

import (
    "bytes"
    "context"
    "crypto/tls"
    "encoding/json"
    "github.com/elastic/go-elasticsearch/v8"
    "github.com/elastic/go-elasticsearch/v8/esapi"
    "log"
    "net/http"
    "strconv"
    "strings"
    "sync"
    "time"
)
var (
    r  map[string]interface{}
    wg sync.WaitGroup
)
func main() {
    cfg := elasticsearch.Config{
        Addresses: []string{
            "https://192.168.1.156:9200",
        },
        Username: "elastic",
        Password: "123456",
        Transport: &http.Transport{
            MaxIdleConnsPerHost:   10,
            ResponseHeaderTimeout: time.Second,
            TLSClientConfig: &tls.Config{
                MinVersion: tls.VersionTLS12,
                InsecureSkipVerify: true,
            },
        },
    }
    es, err := elasticsearch.NewClient(cfg)
    res, err := es.Info()
    if err != nil {
        log.Fatalf("Error getting response: %s", err)
    }
    defer res.Body.Close()
    // Check response status
    if res.IsError() {
        log.Fatalf("Error: %s", res.String())
    }
    // Deserialize the response into a map.
    if err := json.NewDecoder(res.Body).Decode(&r); err != nil {
        log.Fatalf("Error parsing the response body: %s", err)
    }
    // Print client and server version numbers.
    log.Printf("Client: %s", elasticsearch.Version)
    log.Printf("Server: %s", r["version"].(map[string]interface{})["number"])
    log.Println(strings.Repeat("~", 37))

    // 2. Index documents concurrently
    //
    for i, title := range []string{"Test One", "Test Two"} {
        wg.Add(1)

        go func(i int, title string) {
            defer wg.Done()

            // Build the request body.
            data, err := json.Marshal(struct{ Title string }{Title: title})
            if err != nil {
                log.Fatalf("Error marshaling document: %s", err)
            }

            // Set up the request object.
            req := esapi.IndexRequest{
                Index:      "test",
                DocumentID: strconv.Itoa(i + 1),
                Body:       bytes.NewReader(data),
                Refresh:    "true",
            }

            // Perform the request with the client.
            res, err := req.Do(context.Background(), es)
            if err != nil {
                log.Fatalf("Error getting response: %s", err)
            }
            defer res.Body.Close()

            if res.IsError() {
                log.Printf("[%s] Error indexing document ID=%d", res.Status(), i+1)
            } else {
                // Deserialize the response into a map.
                var r map[string]interface{}
                if err := json.NewDecoder(res.Body).Decode(&r); err != nil {
                    log.Printf("Error parsing the response body: %s", err)
                } else {
                    // Print the response status and indexed document version.
                    log.Printf("[%s] %s; version=%d", res.Status(), r["result"], int(r["_version"].(float64)))
                }
            }
        }(i, title)
    }
    wg.Wait()

    log.Println(strings.Repeat("-", 37))

    // 3. Search for the indexed documents
    //
    // Build the request body.
    var buf bytes.Buffer
    query := map[string]interface{}{
        "query": map[string]interface{}{
            "match": map[string]interface{}{
                "title": "test",
            },
        },
    }
    if err := json.NewEncoder(&buf).Encode(query); err != nil {
        log.Fatalf("Error encoding query: %s", err)
    }

    // Perform the search request.
    res, err = es.Search(
        es.Search.WithContext(context.Background()),
        es.Search.WithIndex("test"),
        es.Search.WithBody(&buf),
        es.Search.WithTrackTotalHits(true),
        es.Search.WithPretty(),
    )
    if err != nil {
        log.Fatalf("Error getting response: %s", err)
    }
    defer res.Body.Close()

    if res.IsError() {
        var e map[string]interface{}
        if err := json.NewDecoder(res.Body).Decode(&e); err != nil {
            log.Fatalf("Error parsing the response body: %s", err)
        } else {
            // Print the response status and error information.
            log.Fatalf("[%s] %s: %s",
                res.Status(),
                e["error"].(map[string]interface{})["type"],
                e["error"].(map[string]interface{})["reason"],
            )
        }
    }

    if err := json.NewDecoder(res.Body).Decode(&r); err != nil {
        log.Fatalf("Error parsing the response body: %s", err)
    }
    // Print the response status, number of results, and request duration.
    log.Printf(
        "[%s] %d hits; took: %dms",
        res.Status(),
        int(r["hits"].(map[string]interface{})["total"].(map[string]interface{})["value"].(float64)),
        int(r["took"].(float64)),
    )
    // Print the ID and document source for each hit.
    for _, hit := range r["hits"].(map[string]interface{})["hits"].([]interface{}) {
        log.Printf(" * ID=%s, %s", hit.(map[string]interface{})["_id"], hit.(map[string]interface{})["_source"])
    }

    log.Println(strings.Repeat("=", 37))
}

運(yùn)行結(jié)果:

2022/12/01 14:22:47 Client: 8.5.0
2022/12/01 14:22:47 Server: 8.5.2
2022/12/01 14:22:47 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
2022/12/01 14:22:48 [201 Created] created; version=1
2022/12/01 14:22:48 [201 Created] created; version=1
2022/12/01 14:22:48 -------------------------------------
2022/12/01 14:22:48 [200 OK] 0 hits; took: 6ms
2022/12/01 14:22:48 =====================================