Etcd服务注册发现实例

package main

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

	clientv3 "go.etcd.io/etcd/client/v3"
)

const (
	// ServiceGoods 商品服务
	ServiceGoods = "goods"
)

// serviceEndpointKeyPrefix 服务入口在 etcd 存储的 key 前缀
var serviceEndpointKeyPrefix = "/service_registry_discovery"

// hostname 主机名
var hostname string

// endpoint 访问入口
var endpoint string

// servicePort 服务端口
var servicePort int = 80

// etcdCfg Etcd配置
var etcdCfg = clientv3.Config{
	Endpoints: []string{
		"http://etcd-1.etcd-headless.devops.svc.cluster.local:2379",
		"http://etcd-2.etcd-headless.devops.svc.cluster.local:2379",
		"http://etcd-3.etcd-headless.devops.svc.cluster.local:2379",
	},
	DialTimeout:          time.Second * 30,
	DialKeepAliveTimeout: time.Second * 30,
	Username:             "root",
	Password:             "90CjPHPRlxw=",
}

// init 初始化
func init() {
	hostname, _ = os.Hostname()
	addrs, err := net.InterfaceAddrs()
	if err != nil {
		fmt.Println(err)
		return
	}
	for _, address := range addrs {
		// 检查ip地址判断是否回环地址
		if ipnet, ok := address.(*net.IPNet); ok && !ipnet.IP.IsLoopback() {
			if ipnet.IP.To4() != nil {
				endpoint = fmt.Sprintf("%s:%d", ipnet.IP.String(), servicePort)
				break
			}
		}
	}
}

func main() {
	// 服务注册
	go func() {
		ServiceRegistry()
	}()
	http.HandleFunc(
		"/", func(w http.ResponseWriter, r *http.Request) {
			_, _ = fmt.Fprint(w, "goods")
		},
	)
	http.HandleFunc("/goods/list", GetGoodsList)
	_ = http.ListenAndServe(fmt.Sprintf(":%d", servicePort), nil)
}

// GetGoodsList 获取商品列表
func GetGoodsList(w http.ResponseWriter, r *http.Request) {
	// 生成订单信息
	body, err := ioutil.ReadAll(r.Body)
	if err != nil {
		fmt.Printf("read body err, %v\n", err)
		return
	}
	params := make(map[string]interface{})
	_ = json.Unmarshal(body, &params)
	var traceID = params["trace_id"]
	var res = map[string]interface{}{
		"trace_id": traceID,
		"message":  "get goods list success",
	}
	b, _ := json.Marshal(res)
	fmt.Printf(
		"trace_id: %v, goods list result: %s\n", traceID, string(b),
	)
	_, _ = fmt.Fprint(w, string(b))
}

// ServiceRegistry 服务注册
func ServiceRegistry() {
	hostname, _ = os.Hostname()
	cli, err := clientv3.New(etcdCfg)
	if err != nil {
		panic(err)
	}
	key := fmt.Sprintf("%s/%s/%s", serviceEndpointKeyPrefix, ServiceGoods, hostname)
	ctx := context.Background()
	// 过期时间: 3秒钟
	ttl := 3
	// 创建租约
	lease, err := cli.Grant(ctx, int64(ttl))
	if err != nil {
		panic(err)
	}
	b, _ := json.Marshal(lease)
	fmt.Printf("grant lease suucess: %s\n", string(b))
	// put kv
	res, err := cli.Put(ctx, key, endpoint, clientv3.WithLease(lease.ID))
	if err != nil {
		panic(err)
	}
	b, _ = json.Marshal(res)
	fmt.Printf("put kv with lease suucess: %s\n", string(b))
	// 保持租约不过期
	klRes, err := cli.KeepAlive(ctx, lease.ID)
	if err != nil {
		panic(err)
	}
	// 监听续约情况
	for v := range klRes {
		b, _ = json.Marshal(v)
		fmt.Printf("keep lease alive suucess: %s\n", string(b))
	}
	fmt.Println("stop keeping lease alive")
}

package main

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"io/ioutil"
	"math/rand"
	"net/http"
	"sync"
	"time"

	clientv3 "go.etcd.io/etcd/client/v3"
)

const (
	// ServiceGoods 商品服务
	ServiceGoods = "goods"
)

// serviceEndpointKeyPrefix 服务入口在 etcd 存储的 key 前缀
var serviceEndpointKeyPrefix = "/service_registry_discovery"

// serviceEndpoints 服务入口列表
var serviceEndpoints = map[string]map[string]string{
	ServiceGoods: {},
}

// 全局服务锁
var serviceLocker = sync.Mutex{}

// etcdCfg Etcd配置
var etcdCfg = clientv3.Config{
	Endpoints: []string{
		"http://etcd-1.etcd-headless.devops.svc.cluster.local:2379",
		"http://etcd-2.etcd-headless.devops.svc.cluster.local:2379",
		"http://etcd-3.etcd-headless.devops.svc.cluster.local:2379",
	},
	DialTimeout:          time.Second * 30,
	DialKeepAliveTimeout: time.Second * 30,
	Username:             "root",
	Password:             "90CjPHPRlxw=",
}

// servicePort 服务端口
var servicePort int = 80

func main() {
	// 监听服务入口
	go func() {
		ServiceDiscovery()
	}()
	http.HandleFunc(
		"/", func(w http.ResponseWriter, r *http.Request) {
			_, _ = fmt.Fprint(w, "gateway")
		},
	)
	http.HandleFunc("/goods/list", GetGoodsList)
	_ = http.ListenAndServe(fmt.Sprintf(":%d", servicePort), nil)
}

// GetGoodsList 获取商品列表
func GetGoodsList(w http.ResponseWriter, r *http.Request) {
	var traceID = time.Now().Unix()
	var res = map[string]interface{}{
		"trace_id": traceID,
		"code":     0,
	}
	client, endpoint, err := GetSvcEndpoin(ServiceGoods)
	fmt.Printf(
		"trace_id: %d, get goods endpoint result: client=%s, endpoint=%s, error=%v\n", traceID, client, endpoint, err,
	)
	if err != nil {
		res["code"] = -1
		res["message"] = err.Error()
		b, _ := json.Marshal(res)
		_, _ = fmt.Fprint(w, string(b))
		return
	}
	url := fmt.Sprintf("http://%s/goods/list", endpoint)
	m := map[string]interface{}{
		"trace_id": time.Now().Unix(),
	}
	body, _ := json.Marshal(m)
	resp, err := http.Post(url, "application/json", bytes.NewReader(body))
	if err != nil {
		w.WriteHeader(http.StatusInternalServerError)
		return
	}
	defer func(Body io.ReadCloser) {
		_ = Body.Close()
	}(resp.Body)
	b, _ := ioutil.ReadAll(resp.Body)
	var jsonResp map[string]interface{}
	_ = json.Unmarshal(b, &jsonResp)
	res["message"] = jsonResp["message"]
	b, _ = json.Marshal(res)
	fmt.Printf(
		"trace_id: %d, get goods list result: %s\n", traceID, string(b),
	)
	_, _ = fmt.Fprint(w, string(b))
}

// ServiceDiscovery 服务发现
func ServiceDiscovery() {
	cli, err := clientv3.New(etcdCfg)
	if err != nil {
		panic(err)
	}
	for k, _ := range serviceEndpoints {
		go func(svc string) {
			ctx := context.Background()
			serviceKey := fmt.Sprintf("%s/%s", serviceEndpointKeyPrefix, svc)
			// 获取当前所有服务入口
			getRes, _ := cli.Get(ctx, serviceKey, clientv3.WithPrefix())
			serviceLocker.Lock()
			for _, v := range getRes.Kvs {
				serviceEndpoints[svc][string(v.Key)] = string(v.Value)
			}
			serviceLocker.Unlock()
			fmt.Printf(
				"[service_endpoint_change] [%s] service %s get endpoints success, %v\n", svc, svc,
				serviceEndpoints[svc],
			)
			ch := cli.Watch(ctx, serviceKey, clientv3.WithPrefix(), clientv3.WithPrevKV())
			for v := range ch {
				for _, v := range v.Events {
					key := string(v.Kv.Key)
					endpoint := string(v.Kv.Value)
					preEndpoint := ""
					if v.PrevKv != nil {
						preEndpoint = string(v.PrevKv.Value)
					}
					switch v.Type {
					// PUT，新增或替换
					case 0:
						serviceLocker.Lock()
						serviceEndpoints[svc][key] = endpoint
						serviceLocker.Unlock()
						fmt.Printf(
							"[service_endpoint_change] service %s put endpoint, key: %s, endpoint: %s\n", svc,
							key, endpoint,
						)
					// DELETE
					case 1:
						serviceLocker.Lock()
						delete(serviceEndpoints[svc], key)
						serviceLocker.Unlock()
						fmt.Printf(
							"[service_endpoint_change] service %s delete endpoint, key: %s, endpoint: %s\n",
							svc, key, preEndpoint,
						)
					}
				}
			}
		}(k)
	}
}

// GetSvcEndpoin 获取服务入口
func GetSvcEndpoin(svc string) (key, endpoint string, err error) {
	endpoints := serviceEndpoints[svc]
	if len(endpoints) == 0 {
		return "", "", errors.New(fmt.Sprintf("%s服务不可用，请稍后再试", svc))
	}
	num := len(endpoints)
	keys := make([]string, num)
	for v := range endpoints {
		keys = append(keys, v)
	}
	randomKey := keys[rand.Intn(len(keys))]
	return randomKey, endpoints[randomKey], nil
}

cat >> /etc/hosts <<EOF
192.168.1.101 etcd-demo.local.com
EOF

apiVersion: v1
kind: Namespace
metadata:
  name: etcd-demo
  annotations:
    scheduler.alpha.kubernetes.io/node-selector: usefulness=business

apiVersion: apps/v1
kind: Deployment
metadata:
  name: gateway
  namespace: etcd-demo
  labels:
    app: gateway
spec:
  # 期望 pod 数量
  replicas: 10
  # 新创建的 pod 在运行指定秒数后才视为运行可用，配合就绪探针可以在滚动升级失败的时候阻止升级，避免部署出错的应用
  minReadySeconds: 4
  strategy:
    rollingUpdate:
      # 滚动升级过程中最多允许超出期望副本数的数量，比如期望3，maxSurge 配置为1，则最多存在4个pod，也可以配置百分比
      maxSurge: 1
      # 滚动升级过程中最多允许存在不可用的 pod 数量，配置为0表示升级过程中所有的 pod 都必须可用，即 pod 挨个替换，也可以配置百分比
      maxUnavailable: 0
  # 匹配器，匹配 pod 的方式
  selector:
    matchLabels:
      app: gateway
  template:
    metadata:
      name: gateway
      labels:
        app: gateway
    spec:
      imagePullSecrets:
        - name: harbor-jormin
      containers:
        - name: gateway
          image: harbor.wcxst.com/etcd-demo/gateway:latest
          # 就绪探针
          readinessProbe:
            # 执行周期，单位：秒
            periodSeconds: 1
            # 初始化延迟，单位：秒
            initialDelaySeconds: 3
            httpGet:
              path: /
              port: 80
---

kind: Service
apiVersion: v1
metadata:
  name: gateway
  namespace: etcd-demo
spec:
  ports:
    - protocol: TCP
      port: 80
      targetPort: 80
  selector:
    app: gateway

---

apiVersion: traefik.containo.us/v1alpha1
kind: IngressRoute
metadata:
  name: gateway
  namespace: etcd-demo
spec:
  entryPoints:
    - web
  routes:
    - match: Host(`etcd-demo.local.com`) && PathPrefix(`/`)
      kind: Rule
      services:
        - name: gateway
          port: 80

apiVersion: apps/v1
kind: Deployment
metadata:
  name: goods
  namespace: etcd-demo
  labels:
    app: goods
spec:
  # 期望 pod 数量
  replicas: 3
  # 新创建的 pod 在运行指定秒数后才视为运行可用，配合就绪探针可以在滚动升级失败的时候阻止升级，避免部署出错的应用
  minReadySeconds: 2
  strategy:
    rollingUpdate:
      # 滚动升级过程中最多允许超出期望副本数的数量，比如期望3，maxSurge 配置为1，则最多存在4个pod，也可以配置百分比
      maxSurge: 1
      # 滚动升级过程中最多允许存在不可用的 pod 数量，配置为0表示升级过程中所有的 pod 都必须可用，即 pod 挨个替换，也可以配置百分比
      maxUnavailable: 0
  # 匹配器，匹配 pod 的方式
  selector:
    matchLabels:
      app: goods
  template:
    metadata:
      name: goods
      labels:
        app: goods
    spec:
      imagePullSecrets:
        - name: harbor-jormin
      containers:
        - name: goods
          image: harbor.wcxst.com/etcd-demo/goods:latest
          # 就绪探针
          readinessProbe:
            # 执行周期，单位：秒
            periodSeconds: 1
            # 初始化延迟，单位：秒
            initialDelaySeconds: 3
            httpGet:
              path: /
              port: 80

kubectl apply -f namespace.yaml

root@master:~# curl http://etcd-demo.local.com/goods/list && echo ''
{"code":0,"message":"get goods list success","trace_id":1639626420}

# 从 goods 服务获取结果，goods endpoint 为 10.233.96.155:80
Dec 16, 2021 @ 11:47:00.532	gateway-5b679b6fb-hpqbc	trace_id: 1639626420, get goods endpoint result: client=/service_registry_discovery/goods/goods-7466dd7d4f-pzmc7, endpoint=10.233.96.155:80, error=<nil>

# 最终响应结果
Dec 16, 2021 @ 11:47:00.535	gateway-5b679b6fb-hpqbc	trace_id: 1639626420, get goods list result: {"code":0,"message":"get goods list success","trace_id":1639626420}

Dec 16, 2021 @ 11:47:00.472	goods-7466dd7d4f-pzmc7	trace_id: 1.63962642e+09, goods list result: {"message":"get goods list success","trace_id":1639626420}