gRPC 与 Protobuf 基础

// 文件: api/types/task/task.proto
syntax = "proto3";

package containerd.types;

import "google/protobuf/timestamp.proto";
import "google/protobuf/any.proto";

// 进程信息
message ProcessInfo {
    // 进程 ID
    uint32 pid = 1;

    // 进程状态
    Status status = 2;

    // 标准输入路径
    string stdin = 3;

    // 标准输出路径
    string stdout = 4;

    // 标准错误路径
    string stderr = 5;

    // 是否分配终端
    bool terminal = 6;

    // 退出状态码
    uint32 exit_status = 7;

    // 退出时间
    google.protobuf.Timestamp exited_at = 8;
}

// 进程状态枚举
enum Status {
    UNKNOWN = 0;
    CREATED = 1;
    RUNNING = 2;
    STOPPED = 3;
    PAUSED = 4;
    PAUSING = 5;
}

// 嵌套消息
message Container {
    string id = 1;
    map<string, string> labels = 2;        // 映射类型
    repeated Mount mounts = 3;              // 数组类型
    google.protobuf.Any spec = 4;           // 任意类型
}

message Mount {
    string type = 1;
    string source = 2;
    string target = 3;
    repeated string options = 4;
}

message Example {
    reserved 2, 15, 9 to 11;
    reserved "foo", "bar";

    string name = 1;
    // 字段 2 已废弃，不要使用
    int32 id = 3;
}

// 生成的结构体
type ProcessInfo struct {
    Pid        uint32               `protobuf:"varint,1,opt,name=pid,proto3"`
    Status     Status               `protobuf:"varint,2,opt,name=status,proto3,enum=containerd.types.Status"`
    Stdin      string               `protobuf:"bytes,3,opt,name=stdin,proto3"`
    Stdout     string               `protobuf:"bytes,4,opt,name=stdout,proto3"`
    Stderr     string               `protobuf:"bytes,5,opt,name=stderr,proto3"`
    Terminal   bool                 `protobuf:"varint,6,opt,name=terminal,proto3"`
    ExitStatus uint32               `protobuf:"varint,7,opt,name=exit_status,json=exitStatus,proto3"`
    ExitedAt   *timestamppb.Timestamp `protobuf:"bytes,8,opt,name=exited_at,json=exitedAt,proto3"`
}

// 序列化
data, err := proto.Marshal(info)

// 反序列化
var info ProcessInfo
err := proto.Unmarshal(data, &info)

// 文件: api/services/tasks/v1/tasks.proto
syntax = "proto3";

package containerd.services.tasks.v1;

import "google/protobuf/empty.proto";
import "google/protobuf/any.proto";

// Tasks 服务定义
service Tasks {
    // 创建任务
    rpc Create(CreateTaskRequest) returns (CreateTaskResponse);

    // 启动任务
    rpc Start(StartRequest) returns (StartResponse);

    // 删除任务
    rpc Delete(DeleteTaskRequest) returns (DeleteResponse);

    // 获取任务状态
    rpc Get(GetRequest) returns (GetResponse);

    // 列出所有任务
    rpc List(ListTasksRequest) returns (ListTasksResponse);

    // 杀死任务
    rpc Kill(KillRequest) returns (google.protobuf.Empty);

    // 在任务中执行命令
    rpc Exec(ExecProcessRequest) returns (google.protobuf.Empty);

    // 等待任务退出
    rpc Wait(WaitRequest) returns (WaitResponse);
}

message CreateTaskRequest {
    string container_id = 1;
    repeated containerd.types.Mount rootfs = 3;
    string stdin = 4;
    string stdout = 5;
    string stderr = 6;
    bool terminal = 7;
    containerd.types.Mount checkpoint = 8;
    google.protobuf.Any options = 9;
    string runtime_path = 10;
}

message CreateTaskResponse {
    string container_id = 1;
    uint32 pid = 2;
}

service Example {
    // Unary RPC - 一请求一响应
    rpc GetInfo(GetRequest) returns (GetResponse);

    // Server Streaming - 一请求多响应
    rpc Subscribe(SubscribeRequest) returns (stream Event);

    // Client Streaming - 多请求一响应
    rpc Upload(stream DataChunk) returns (UploadResponse);

    // Bidirectional Streaming - 双向流
    rpc Chat(stream Message) returns (stream Message);
}

// 服务接口（由 protoc 生成）
type TasksServer interface {
    Create(context.Context, *CreateTaskRequest) (*CreateTaskResponse, error)
    Start(context.Context, *StartRequest) (*StartResponse, error)
    Delete(context.Context, *DeleteTaskRequest) (*DeleteResponse, error)
    // ...
}

// 服务实现
type tasksService struct {
    UnimplementedTasksServer  // 嵌入未实现的服务
    runtime PlatformRuntime
}

func (s *tasksService) Create(ctx context.Context, req *CreateTaskRequest) (*CreateTaskResponse, error) {
    // 实现创建任务逻辑
    task, err := s.runtime.Create(ctx, req.ContainerId, opts)
    if err != nil {
        return nil, err
    }
    return &CreateTaskResponse{
        ContainerId: req.ContainerId,
        Pid:         task.Pid(),
    }, nil
}

// 注册服务
func RegisterTasksServer(s *grpc.Server, srv TasksServer) {
    s.RegisterService(&_Tasks_serviceDesc, srv)
}

// 创建连接
conn, err := grpc.Dial("unix:///run/containerd/containerd.sock",
    grpc.WithTransportCredentials(insecure.NewCredentials()),
)
if err != nil {
    log.Fatal(err)
}
defer conn.Close()

// 创建客户端
client := tasksv1.NewTasksClient(conn)

// 调用服务
resp, err := client.Create(context.Background(), &tasksv1.CreateTaskRequest{
    ContainerId: "my-container",
    // ...
})
if err != nil {
    log.Fatal(err)
}
fmt.Printf("Created task with PID: %d\n", resp.Pid)

// 服务端
server, err := ttrpc.NewServer()
if err != nil {
    log.Fatal(err)
}

// 注册服务
tasksapi.RegisterTasksService(server, &taskService{})

// 监听 Unix Socket
listener, err := net.Listen("unix", "/run/containerd/shim.sock")
if err != nil {
    log.Fatal(err)
}

// 启动服务
server.Serve(ctx, listener)

// 客户端
conn, err := net.Dial("unix", "/run/containerd/shim.sock")
if err != nil {
    log.Fatal(err)
}

client := ttrpc.NewClient(conn)
tasksClient := tasksapi.NewTasksClient(client)

// 调用
resp, err := tasksClient.Create(ctx, &tasksapi.CreateTaskRequest{...})

// api/services/images/v1/images.proto
service Images {
    rpc Get(GetImageRequest) returns (GetImageResponse);
    rpc List(ListImagesRequest) returns (ListImagesResponse);
    rpc Create(CreateImageRequest) returns (CreateImageResponse);
    rpc Update(UpdateImageRequest) returns (UpdateImageResponse);
    rpc Delete(DeleteImageRequest) returns (google.protobuf.Empty);
}

message Image {
    string name = 1;
    map<string, string> labels = 2;
    containerd.types.Descriptor target = 3;
    google.protobuf.Timestamp created_at = 7;
    google.protobuf.Timestamp updated_at = 8;
}

import (
    "github.com/containerd/containerd/v2"
    "github.com/containerd/containerd/v2/namespaces"
)

// 创建客户端
client, err := containerd.New("/run/containerd/containerd.sock")
if err != nil {
    log.Fatal(err)
}
defer client.Close()

// 设置命名空间
ctx := namespaces.WithNamespace(context.Background(), "default")

// 拉取镜像
image, err := client.Pull(ctx, "docker.io/library/nginx:latest",
    containerd.WithPullUnpack,
)
if err != nil {
    log.Fatal(err)
}

// 创建容器
container, err := client.NewContainer(ctx, "my-nginx",
    containerd.WithImage(image),
    containerd.WithNewSnapshot("my-nginx-snapshot", image),
    containerd.WithNewSpec(oci.WithImageConfig(image)),
)
if err != nil {
    log.Fatal(err)
}

// 创建任务
task, err := container.NewTask(ctx, cio.NewCreator(cio.WithStdio))
if err != nil {
    log.Fatal(err)
}

// 启动任务
err = task.Start(ctx)

# api/buf.yaml
version: v1
deps:
  - buf.build/googleapis/googleapis
  - buf.build/protocolbuffers/wellknowntypes

# api/buf.gen.yaml
version: v1
plugins:
  - plugin: go
    out: .
    opt:
      - paths=source_relative
  - plugin: go-grpc
    out: .
    opt:
      - paths=source_relative
  - plugin: go-ttrpc
    out: .
    opt:
      - paths=source_relative

import "google.golang.org/grpc/grpclog"

// 设置日志级别
grpclog.SetLoggerV2(grpclog.NewLoggerV2WithVerbosity(
    os.Stderr, os.Stderr, os.Stderr, 2,
))