gistfile1.txt

package main

import (
    "context"
    "fmt"
    "log"
    "net"
    "net/url"
    "os"
    "path"
    "runtime/trace"
    "sync/atomic"
    "time"

    "github.com/coreos/etcd/clientv3"
    "github.com/coreos/etcd/pkg/transport"
    grpcprom "github.com/grpc-ecosystem/go-grpc-prometheus"
    "google.golang.org/grpc"

    utilnet "k8s.io/apimachinery/pkg/util/net"
    "k8s.io/apimachinery/pkg/util/wait"
    "k8s.io/apiserver/pkg/server/egressselector"
    "k8s.io/apiserver/pkg/storage/storagebackend"
)

// The short keepalive timeout and interval have been chosen to aggressively
// detect a failed etcd server without introducing much overhead.
const keepaliveTime = 30 * time.Second
const keepaliveTimeout = 10 * time.Second

// dialTimeout is the timeout for failing to establish a connection.
// It is set to 20 seconds as times shorter than that will cause TLS connections to fail
// on heavily loaded arm64 CPUs (issue #64649)
const dialTimeout = 20 * time.Second

func main()
    // NOTE: to see debug output you must
    // export ETCD_CLIENT_DEBUG=1
    
    f, err := os.Create("trace.out")
    if err != nil {
        panic(err)
    }   
    defer f.Close()

    err = trace.Start(f)
    if err != nil {
        panic(err)
    }   
    defer trace.Stop()
    projectDir, err := os.Getwd()
    if err != nil {
        log.Fatal(err)
    }   

    t := storagebackend.TransportConfig{
        ServerList:    []string{"https://node-1.hexfusion.local:2379", "https://node-2.hexfusion.local:2379", "https://node-3.hexfusion.local:2379"},
        KeyFile:       fmt.Sprintf("%s/fixtures/certs/client/client-key.pem", projectDir),
        CertFile:      fmt.Sprintf("%s/fixtures/certs/client/client.pem", projectDir),
        TrustedCAFile: fmt.Sprintf("%s/fixtures/certs/ca/ca.pem", projectDir),
    }   
    c, err := newETCD3Client(t)
    if err != nil {
        log.Fatal(err)
    }   
    defer c.Close()

    rch := c.Watch(context.Background(), "foo")
    for wresp := range rch {
        for _, ev := range wresp.Events {
            fmt.Printf("%s %q : %q\n", ev.Type, ev.Kv.Key, ev.Kv.Value)
        }
    }   
}

func newETCD3Client(c storagebackend.TransportConfig) (*clientv3.Client, error) {
    tlsInfo := transport.TLSInfo{
        CertFile:      c.CertFile,
        KeyFile:       c.KeyFile,
        TrustedCAFile: c.TrustedCAFile,
    }   
    tlsConfig, err := tlsInfo.ClientConfig()
    if err != nil {
        return nil, err
    }
    // NOTE: Client relies on nil tlsConfig
    // for non-secure connections, update the implicit variable
    if len(c.CertFile) == 0 && len(c.KeyFile) == 0 && len(c.TrustedCAFile) == 0 {
        tlsConfig = nil
    }
    networkContext := egressselector.Etcd.AsNetworkContext()
    var egressDialer utilnet.DialFunc
    if c.EgressLookup != nil {
        egressDialer, err = c.EgressLookup(networkContext)
        if err != nil {
            return nil, err
        }
    }

    dialOptions := []grpc.DialOption{
        grpc.WithBlock(), // block until the underlying connection is up
        grpc.WithUnaryInterceptor(grpcprom.UnaryClientInterceptor),
        grpc.WithStreamInterceptor(grpcprom.StreamClientInterceptor),
    }
    if egressDialer != nil {
        dialer := func(ctx context.Context, addr string) (net.Conn, error) {
            u, err := url.Parse(addr)
            if err != nil {
                return nil, err
            }
            return egressDialer(ctx, "tcp", u.Host)
        }
        dialOptions = append(dialOptions, grpc.WithContextDialer(dialer))
    }
    cfg := clientv3.Config{
        DialTimeout:          dialTimeout,
        DialKeepAliveTime:    keepaliveTime,
        DialKeepAliveTimeout: keepaliveTimeout,
        DialOptions:          dialOptions,
        Endpoints:            c.ServerList,
        TLS:                  tlsConfig,
    }

    return clientv3.New(cfg)
}