하고 싶은 일

발판 서버를 통해 (ssh) 프라이빗 서브넷의 RDS(MySQL)에 연결하는 것과 마찬가지로 VPC 엔드포인트의 Elasticsearch Service에도 발판 서버를 통해 연결하는 것이 목표입니다.

ssh 및 curl 명령을 사용하면 다음과 같이 쉽게 로컬에서 연결할 수 있지만 이번에는 Go 프로그램으로 연결하는 것을 목표로 합니다.

ssh -i <path/to/private-key> <username>@<hostname> curl -s '<ES_ENDPOINT>/_cat/indices?format=json&pretty'

전제

다음과 같은 아키텍처를 가정합니다. 보안 그룹은 좋은 느낌으로 설정되어 있다고 가정합니다.

구현

Elasticsearch의 클라이언트 라이브러리는 Elastic의 공식 elastic/go-elasticsearch을 사용합니다.

구현의 포인트는 http.RoundTripper (http.Transport)의 Dial에 SSH Client의 Dial을 이용하는 것입니다.

package main

import (
    "context"
    "encoding/json"
    "fmt"
    "io"
    "io/ioutil"
    "log"
    "net"
    "net/http"
    "os"
    "time"

    "github.com/elastic/go-elasticsearch/v8"
    "github.com/elastic/go-elasticsearch/v8/esapi"
    "golang.org/x/crypto/ssh"
)

func main() {
    log.SetFlags(log.LstdFlags | log.Lshortfile)

    var (
        sshUser       = os.Getenv("SSH_USER")
        sshHost       = os.Getenv("SSH_HOST")
        sshPort       = os.Getenv("SSH_PORT")
        sshPrivateKey = os.Getenv("SSH_PRIVATE_KEY")
        esEndpoint    = os.Getenv("ES_ENDPOINT")
    )

    // ------------------------------
    // 秘密鍵ファイルの読み込み
    // ------------------------------
    b, err := ioutil.ReadFile(sshPrivateKey)
    if err != nil {
        log.Fatal(err)
    }

    signer, err := ssh.ParsePrivateKey(b)
    if err != nil {
        log.Fatal(err)
    }

    // ------------------------------
    // SSH クライアントの生成
    // ------------------------------
    sshConf := ssh.ClientConfig{
        User: sshUser,
        Auth: []ssh.AuthMethod{
            ssh.PublicKeys(signer),
        },
        HostKeyCallback: func(hostname string, remote net.Addr, key ssh.PublicKey) error {
            return nil
        },
        Timeout: 10 * time.Second,
    }

    sshClient, err := ssh.Dial("tcp", net.JoinHostPort(sshHost, sshPort), &sshConf)
    if err != nil {
        log.Fatal(err)
    }
    defer sshClient.Close()

    // ------------------------------
    // Elasticsearch クライアントの生成
    // ------------------------------
    esConf := elasticsearch.Config{
        Addresses: []string{esEndpoint},
        Transport: &http.Transport{
            Proxy:               http.ProxyFromEnvironment,
            Dial:                sshClient.Dial, // ここで SSH Client を利用
            TLSHandshakeTimeout: 10 * time.Second,
        },
    }

    es, err := elasticsearch.NewClient(esConf)
    if err != nil {
        log.Fatal(err)
    }

    // ------------------------------
    // リクエストを実行 (/_cat/indices)
    // ------------------------------
    req := esapi.CatIndicesRequest{
        Format: "json",
        Pretty: true,
    }

    ctx := context.Background()

    resp, err := req.Do(ctx, es)
    if err != nil {
        log.Fatal(err)
    }
    defer resp.Body.Close()

    // ------------------------------
    // レスポンスを解析
    // ------------------------------
    if resp.IsError() {
        log.Fatal(resp.String())
    }

    body := io.TeeReader(resp.Body, os.Stdout) // debug

    var r []map[string]interface{}
    if err := json.NewDecoder(body).Decode(&r); err != nil {
        log.Fatal(err)
    }

    for i, obj := range r {
        fmt.Printf("\n[#%d]\n", i)
        for k, v := range obj {
            fmt.Println(k, v)
        }
    }
}