installer.md

概述

设计目标

测试

容器测试

容器测试1

容器编译运行 installer-manager

# mysql 依赖
docker run --name mysql  -e MYSQL_DATABASE=manager -e MYSQL_ROOT_PASSWORD=123.com/456 -p 127.0.0.1:3306:3306 -d mysql:5.7

docker build -t installer .

docker run --net=host -e SC_IP=127.0.0.1 -d -p 8098:8098 --name installer installer
# docker run --net=host -e SC_IP=172.18.0.3 -d -p 8098:8098 --name installer installer

容器测试2

准备 se 机器

docker build -f test/se/Dockerfile -t se .

docker run -d -p 22222:22222 --privileged --name se -v /sys/fs/cgroup:/sys/fs/cgroup:ro  -v /run/dbus/system_bus_socket:/run/dbus/system_bus_socket:ro -v /run/dbus:/run/dbus  --cap-add SYS_ADMIN se && docker exec se /scripts/startup.sh

容器测试3

docker compose 将 mysql installer-manager se 都启动，使 installer-manager 对执行测试用例

docker build -t installer . && docker-compose -f test/e2e/base.yml up -d

# 停止 docker-compose  -f test/e2e/base.yml down

容器测试4

package sccore

import (
	"context"
	"crypto/tls"
	"crypto/x509"
	"encoding/json"
	"fmt"
	"io"
	"log"
	"strings"
	"sync"
	"testing"
	"time"

	"google.golang.org/grpc/credentials"

	"git.ouryun.cn/base/go-kit/pkg/ca"
	mdv1 "git.ouryun.cn/base/proto-store/pkg/api/common/metadata/v1"
	"git.ouryun.cn/base/proto-store/pkg/client"
	proto_common "git.ouryun.cn/base/proto-store/pkg/common"
	"git.ouryun.cn/base/proto-store/pkg/interfaces"
	"git.ouryun.cn/base/proto-store/pkg/options"
	"git.ouryun.cn/base/proto-store/pkg/proto_store"
	"git.ouryun.cn/base/proto-store/pkg/types"
)

const (
	managerClientReconnectTimeout = time.Second * 3
	scCoreAddress                 = "sc.srhino.svc.local"
	scCorePort                    = 20111
)
const (
	ExecCommand = "/v1/agent/cmd"
	UploadFile  = "/v1/agent/upload"
)

var (
	managerClientInstance *managerClient
	mgntClientMu          = sync.Mutex{}
)

type ExitError struct {
	code int

	// Stderr holds a subset of the standard error output from the
	// Cmd.Output method if standard error was not otherwise being
	// collected.
	//
	// If the error output is long, Stderr may contain only a prefix
	// and suffix of the output, with the middle replaced with
	// text about the number of omitted bytes.
	//
	// Stderr is provided for debugging, for inclusion in error messages.
	// Users with other needs should redirect Cmd.Stderr as needed.
	Stderr []byte
}

func (e *ExitError) Error() string {
	return fmt.Sprintf("exit with status %d, err info: %s", e.code, e.Stderr)
}

type RPCError struct {
	statusCode proto_common.StatusCode
	err        error
}

func (e *RPCError) Error() string {
	return fmt.Sprintf("rpc error %v, status:%v", e.err.Error(), e.statusCode)
}

func (e *RPCError) Unwrap() error {
	return e.err
}

type managerClient struct {
	interfaces.CoreClient
}

type engineClient struct {
	mu          sync.RWMutex
	cli         *managerClient
	config      *EngineClientConfig
	nodeClients map[string]*NodeClient
}

type NodeClient struct {
	engineCli *engineClient
	nodeId    string
}

type EngineClientConfig struct {
	SN string
	// zero is no timeout
	Timeout time.Duration
}

type CmdRequest struct {
	Name string   `json:"name"`
	Args []string `json:"args"`
}

type CmdResponse struct {
	ExitCode int    `json:"exitCode"`
	Stdout   string `json:"stdout"`
	Stderr   string `json:"stderr"`
}

func (c *NodeClient) callExecCommand(name string, args ...string) (*CmdResponse, error) {
	reqBody := &CmdRequest{
		Name: name,
		Args: args,
	}

	code, bodyBytes, err := c.Post(ExecCommand, reqBody)
	if err != nil {
		rpcerr := &RPCError{
			err: err,
		}
		return nil, rpcerr
	}

	if code == proto_common.StatusServiceUnavailable {
		rpcerr := &RPCError{
			statusCode: code,
			err:        proto_common.ErrNotFoundClient,
		}
		return nil, rpcerr
	} else if code != proto_common.StatusOK {
		rpcerr := &RPCError{
			statusCode: code,
		}
		return nil, rpcerr
	}

	resp := &CmdResponse{}
	if err := json.Unmarshal(bodyBytes, resp); err != nil {
		return nil, err
	}

	return resp, nil
}

func (c *NodeClient) Output(name string, args ...string) ([]byte, error) {
	resp, err := c.callExecCommand(name, args...)
	if err != nil {
		return nil, err
	}

	if resp.ExitCode != 0 {
		return nil, &ExitError{code: resp.ExitCode, Stderr: []byte(resp.Stderr)}
	}

	return []byte(resp.Stdout), nil
}

func (c *NodeClient) CombinedOutput(name string, args ...string) ([]byte, error) {
	resp, err := c.callExecCommand(name, args...)
	if err != nil {
		return nil, err
	}

	if resp.ExitCode != 0 {
		return nil, &ExitError{code: resp.ExitCode, Stderr: []byte(resp.Stderr)}
	}

	return []byte(resp.Stdout + resp.Stderr), nil
}

func (c *NodeClient) Upload(local io.Reader, remotePath string) error {
	// 构造请求
	protoCtx := proto_store.NewRequestContext()
	dst := &mdv1.Metadata_Info{
		Engine:    &mdv1.Engine{SerialNumber: c.engineCli.config.SN},
		Component: types.COMPONENT_TAG_WEBSERVER,
		Node:      &mdv1.Node{Id: c.nodeId},
	}

	req := protoCtx.SetUrl(UploadFile).SetDst(dst)
	req.SetParams("path", remotePath)
	req.SetFileReader(local)
	req.SetMethod(mdv1.Method_POST.String())
	resp, err := c.engineCli.cli.FileStream(req)
	if err != nil {
		rpcerr := &RPCError{
			err: err,
		}
		return rpcerr
	}
	defer resp.Response().Body.Close()

	code := resp.Response().StatusCode
	if code == proto_common.StatusServiceUnavailable {
		rpcerr := &RPCError{
			statusCode: code,
			err:        proto_common.ErrNotFoundClient,
		}
		return rpcerr
	} else if code != proto_common.StatusOK {
		rpcerr := &RPCError{
			statusCode: code,
		}
		return rpcerr
	}

	return nil
}

func (c *NodeClient) Close() error {
	_ = c.engineCli.nodeClose(c.nodeId)
	c.engineCli = nil
	c.nodeId = ""
	return nil
}

func (c *NodeClient) String() string {
	return fmt.Sprintf("node cliet for engine:[%v], node:[%v]", c.engineCli.config.SN, c.nodeId)
}

// Hello 用于检查节点的webserver服务是否可用
func (c *NodeClient) Hello() error {
	outputBytes, err := c.Output("echo", "hello")
	if err != nil {
		return err
	}
	output := strings.TrimSpace(string(outputBytes))
	if output != "hello" {
		return fmt.Errorf("unexpect result")
	}
	return nil
}

func (c *NodeClient) Post(api string, body any) (proto_common.StatusCode, []byte, error) {
	return c.engineCli.Post(c.nodeId, api, body)
}
func (c *NodeClient) Get(api string) (proto_common.StatusCode, []byte, error) {
	return c.engineCli.Get(c.nodeId, api)
}

// @parm host 用于定位具体的节点，应该传递nodeUID
// @param api 由于这个参数仅包含api路径，不包含能用于定位对端服务的host，所以不起名为url
func (c *engineClient) Post(host string, api string, body any) (proto_common.StatusCode, []byte, error) {
	// 构造请求
	protoCtx := proto_store.NewRequestContext()
	dst := &mdv1.Metadata_Info{
		Engine:    &mdv1.Engine{SerialNumber: c.config.SN},
		Component: types.COMPONENT_TAG_WEBSERVER,
	}
	if host != "" {
		dst.Node = &mdv1.Node{Id: host}
	}
	req := protoCtx.SetUrl(api).SetDst(dst)
	if c.config.Timeout != 0 {
		req.SetTimeout(c.config.Timeout)
	}
	// 设置body
	if body != nil {
		if err := req.SetRequestBody(body); err != nil {
			return 0, nil, err
		}
	}
	// 发送请求
	resp, err := c.cli.Post(req)
	if err != nil {
		return 0, nil, err
	}

	// 资源回收
	defer resp.Body.Close()

	// 读取响应
	respBody, err := io.ReadAll(resp.Body)
	if err != nil {
		return 0, nil, err
	}

	return resp.StatusCode, respBody, nil
}

func (c *engineClient) Get(host string, api string) (proto_common.StatusCode, []byte, error) {
	// 构造请求
	protoCtx := proto_store.NewRequestContext()
	dst := &mdv1.Metadata_Info{
		Engine:    &mdv1.Engine{SerialNumber: c.config.SN},
		Component: types.COMPONENT_TAG_WEBSERVER,
	}
	if host != "" {
		dst.Node = &mdv1.Node{Id: host}
	}
	req := protoCtx.SetUrl(api).SetDst(dst)
	if c.config.Timeout != 0 {
		req.SetTimeout(c.config.Timeout)
	}

	// 发送请求
	resp, err := c.cli.Get(req)
	if err != nil {
		return 0, nil, err
	}

	// 资源回收
	defer resp.Body.Close()

	// 读取响应
	respBody, err := io.ReadAll(resp.Body)
	if err != nil {
		return 0, nil, err
	}

	return resp.StatusCode, respBody, nil
}

func (c *engineClient) NodeClient(id string) *NodeClient {
	if id == "" {
		panic("node id should not be empty")
	}

	nodeCli, ok := c.nodeClients[id]
	if !ok {
		nodeCli = &NodeClient{
			engineCli: c,
			nodeId:    id,
		}
		c.nodeClients[id] = nodeCli
	}

	return nodeCli
}

func (c *engineClient) nodeClose(id string) error {
	delete(c.nodeClients, id)
	return nil
}

func (c *engineClient) Close() error {
	c.mu.Lock()
	defer c.mu.Unlock()

	// 关闭node client
	keys := make([]string, 0, len(c.nodeClients))
	for k := range c.nodeClients {
		keys = append(keys, k)
	}

	for _, key := range keys {
		_ = c.nodeClose(key)
	}

	// 关闭连接
	// c.cli.Stop()

	return nil
}

func newManagerClient(tlsConfig *tls.Config) *managerClient {
	creds := credentials.NewTLS(tlsConfig)
	cli, err := client.NewClientService(context.Background(), options.ServiceOption{
		Address:   scCoreAddress,
		Port:      scCorePort,
		Component: types.COMPONENT_TAG_UNKNOWN,
		Creds:     creds,
	})
	if err != nil {
		panic(err)
	}
	if err := cli.Connect(); err != nil {
		// logger.Logger.Warnf("connect to sc-core failed:%v", err)
		return nil
	}
	go func() {
		for {
			err := cli.Start()
			if err != nil {
				// logger.Logger.Errorf("manager core client start failed:%v", err.Error())
			}
			time.Sleep(managerClientReconnectTimeout)
			if err := cli.Connect(); err != nil {
				// logger.Logger.Warnf("connect to sc-core failed:%v", err)
			}
		}
	}()

	return &managerClient{CoreClient: cli}
}

func getManagerClient() *managerClient {
	mgntClientMu.Lock()
	defer mgntClientMu.Unlock()
	if managerClientInstance == nil {
		clientCert, err := tls.X509KeyPair(ca.Generate("Manager"))
		if err != nil {
			log.Fatalf("Failed to load client certificate: %v", err)
		}
		caCert, _ := ca.RootCa()
		caCertPool := x509.NewCertPool()
		caCertPool.AppendCertsFromPEM(caCert)
		_ = clientCert
		clientTLSConfig := &tls.Config{
			// ServerName:   "sc.srhino.svc.local",
			// Certificates: []tls.Certificate{clientCert},
			// RootCAs:      caCertPool, // 信任服务端证书的根 CA
			InsecureSkipVerify: true,
		}

		managerClientInstance = newManagerClient(clientTLSConfig)
	}

	return managerClientInstance
}

// 当manager与sc-core无法通信时才会返回错误
func NewEngineClient(config *EngineClientConfig) (*engineClient, error) {
	cli := getManagerClient()
	if cli == nil {
		return nil, fmt.Errorf("get manager client failed")
	}

	return &engineClient{
		cli:         cli,
		config:      config,
		nodeClients: make(map[string]*NodeClient),
	}, nil
}

func TestEngineClient(t *testing.T) {
	engineCli, err := NewEngineClient(&EngineClientConfig{SN: "SN123"})
	if err != nil {
		t.Error(err)
	}
	defer engineCli.Close()
	nodeCli := engineCli.NodeClient("69ea20be-8be2-4432-8c0a-288af1c9deae")
	if err := nodeCli.Hello(); err != nil {
		t.Error(err)
	}
}