Kubernetes容器运行时 — CRI接口

// Runtime service defines the public APIs for remote container runtimes
service RuntimeService {
    // Version returns the runtime name, runtime version, and runtime API version.
    rpc Version(VersionRequest) returns (VersionResponse) {}


    // RunPodSandbox creates and starts a pod-level sandbox. Runtimes must ensure
    // the sandbox is in the ready state on success.
    rpc RunPodSandbox(RunPodSandboxRequest) returns (RunPodSandboxResponse) {}
    // StopPodSandbox stops any running process that is part of the sandbox and
    // reclaims network resources (e.g., IP addresses) allocated to the sandbox.
    // If there are any running containers in the sandbox, they must be forcibly
    // terminated.
    // This call is idempotent, and must not return an error if all relevant
    // resources have already been reclaimed. kubelet will call StopPodSandbox
    // at least once before calling RemovePodSandbox. It will also attempt to
    // reclaim resources eagerly, as soon as a sandbox is not needed. Hence,
    // multiple StopPodSandbox calls are expected.
    rpc StopPodSandbox(StopPodSandboxRequest) returns (StopPodSandboxResponse) {}
    // RemovePodSandbox removes the sandbox. If there are any running containers
    // in the sandbox, they must be forcibly terminated and removed.
    // This call is idempotent, and must not return an error if the sandbox has
    // already been removed.
    rpc RemovePodSandbox(RemovePodSandboxRequest) returns (RemovePodSandboxResponse) {}
    // PodSandboxStatus returns the status of the PodSandbox. If the PodSandbox is not
    // present, returns an error.
    rpc PodSandboxStatus(PodSandboxStatusRequest) returns (PodSandboxStatusResponse) {}
    // ListPodSandbox returns a list of PodSandboxes.
    rpc ListPodSandbox(ListPodSandboxRequest) returns (ListPodSandboxResponse) {}


    // CreateContainer creates a new container in specified PodSandbox
    rpc CreateContainer(CreateContainerRequest) returns (CreateContainerResponse) {}
    // StartContainer starts the container.
    rpc StartContainer(StartContainerRequest) returns (StartContainerResponse) {}
    // StopContainer stops a running container with a grace period (i.e., timeout).
    // This call is idempotent, and must not return an error if the container has
    // already been stopped.
    // TODO: what must the runtime do after the grace period is reached?
    rpc StopContainer(StopContainerRequest) returns (StopContainerResponse) {}
    // RemoveContainer removes the container. If the container is running, the
    // container must be forcibly removed.
    // This call is idempotent, and must not return an error if the container has
    // already been removed.
    rpc RemoveContainer(RemoveContainerRequest) returns (RemoveContainerResponse) {}
    // ListContainers lists all containers by filters.
    rpc ListContainers(ListContainersRequest) returns (ListContainersResponse) {}
    // ContainerStatus returns status of the container. If the container is not
    // present, returns an error.
    rpc ContainerStatus(ContainerStatusRequest) returns (ContainerStatusResponse) {}
    // UpdateContainerResources updates ContainerConfig of the container.
    rpc UpdateContainerResources(UpdateContainerResourcesRequest) returns (UpdateContainerResourcesResponse) {}
    // ReopenContainerLog asks runtime to reopen the stdout/stderr log file
    // for the container. This is often called after the log file has been
    // rotated. If the container is not running, container runtime can choose
    // to either create a new log file and return nil, or return an error.
    // Once it returns error, new container log file MUST NOT be created.
    rpc ReopenContainerLog(ReopenContainerLogRequest) returns (ReopenContainerLogResponse) {}


    // ExecSync runs a command in a container synchronously.
    rpc ExecSync(ExecSyncRequest) returns (ExecSyncResponse) {}
    // Exec prepares a streaming endpoint to execute a command in the container.
    rpc Exec(ExecRequest) returns (ExecResponse) {}
    // Attach prepares a streaming endpoint to attach to a running container.
    rpc Attach(AttachRequest) returns (AttachResponse) {}
    // PortForward prepares a streaming endpoint to forward ports from a PodSandbox.
    rpc PortForward(PortForwardRequest) returns (PortForwardResponse) {}


    // ContainerStats returns stats of the container. If the container does not
    // exist, the call returns an error.
    rpc ContainerStats(ContainerStatsRequest) returns (ContainerStatsResponse) {}
    // ListContainerStats returns stats of all running containers.
    rpc ListContainerStats(ListContainerStatsRequest) returns (ListContainerStatsResponse) {}


    // UpdateRuntimeConfig updates the runtime configuration based on the given request.
    rpc UpdateRuntimeConfig(UpdateRuntimeConfigRequest) returns (UpdateRuntimeConfigResponse) {}


    // Status returns the status of the runtime.
    rpc Status(StatusRequest) returns (StatusResponse) {}
}


// ImageService defines the public APIs for managing images.
service ImageService {
    // ListImages lists existing images.
    rpc ListImages(ListImagesRequest) returns (ListImagesResponse) {}
    // ImageStatus returns the status of the image. If the image is not
    // present, returns a response with ImageStatusResponse.Image set to
    // nil.
    rpc ImageStatus(ImageStatusRequest) returns (ImageStatusResponse) {}
    // PullImage pulls an image with authentication config.
    rpc PullImage(PullImageRequest) returns (PullImageResponse) {}
    // RemoveImage removes the image.
    // This call is idempotent, and must not return an error if the image has
    // already been removed.
    rpc RemoveImage(RemoveImageRequest) returns (RemoveImageResponse) {}
    // ImageFSInfo returns information of the filesystem that is used to store images.
    rpc ImageFsInfo(ImageFsInfoRequest) returns (ImageFsInfoResponse) {}
}

  --container-runtime=remote \
  --container-runtime-endpoint=/var/run/crio/crio.sock \
  --runtime-request-timeout=5m \
  --image-service-endpoint=/var/run/crio/crio.sock \

[root@k8s-test09 ~]# kubectl get nodes -o wide
NAME         STATUS   ROLES    AGE   VERSION   INTERNAL-IP      EXTERNAL-IP   OS-IMAGE                KERNEL-VERSION                 CONTAINER-RUNTIME
k8s-test09   Ready    <none>   24d   v1.19.2   172.31.250.215   <none>        CentOS Linux 8 (Core)   4.18.0-193.14.2.el8_2.x86_64   cri-o://1.19.0
k8s-test10   Ready    <none>   24d   v1.19.2   172.31.250.214   <none>        CentOS Linux 8 (Core)   4.18.0-193.14.2.el8_2.x86_64   docker://19.3.13
[root@k8s-test09 ~]# 

package main

import (
    "context"
    "fmt"
    "log"
    "net"
    "net/url"
    "os"
    "time"

    "github.com/pkg/errors"
    "google.golang.org/grpc"
    pb "k8s.io/cri-api/pkg/apis/runtime/v1alpha2"
)

const (
    // unixProtocol is the network protocol of unix socket.
    unixProtocol = "unix"
)

var (
    // RuntimeEndpoint is CRI server runtime endpoint
    RuntimeEndpoint = []string{"unix:///var/run/dockershim.sock", "unix:///run/crio/crio.sock", "unix:///run/containerd/containerd.sock"}
    //RuntimeEndpoint = []string{"unix:///run/crio/crio.sock"}
    // Timeout  of connecting to server (default: 10s)
    Timeout = 10 * time.Second
)

func getConnection(endPoints []string) (*grpc.ClientConn, error) {
    if endPoints == nil || len(endPoints) == 0 {
        return nil, fmt.Errorf("endpoint is not set")
    }
    endPointsLen := len(endPoints)
    var conn *grpc.ClientConn
    for indx, endPoint := range endPoints {
        _, sock, _ := parseEndpoint(endPoint)
        if isExist(sock)==false{
            log.Printf("sock: %s No such file or directory\n",sock)
            continue
        }
        log.Printf("connect using endpoint '%s' with '%s' timeout", endPoint, Timeout)
        addr, dialer, err := GetAddressAndDialer(endPoint)
        if err != nil {
            if indx == endPointsLen-1 {
                return nil, err
            }
            log.Fatal(err)
            continue
        }
        conn, err = grpc.Dial(addr, grpc.WithInsecure(), grpc.WithBlock(), grpc.WithTimeout(Timeout), grpc.WithContextDialer(dialer))
        if err != nil {
            errMsg := errors.Wrapf(err, "connect endpoint '%s', make sure you are running as root and the endpoint has been started", endPoint)
            if indx == endPointsLen-1 {
                return nil, errMsg
            }
            log.Println(errMsg)
        } else {
            log.Printf("connected successfully using endpoint: %s", endPoint)
            break
        }
    }
    return conn, nil
}

// GetAddressAndDialer returns the address parsed from the given endpoint and a context dialer.
func GetAddressAndDialer(endpoint string) (string, func(ctx context.Context, addr string) (net.Conn, error), error) {
    protocol, addr, err := parseEndpointWithFallbackProtocol(endpoint, unixProtocol)
    if err != nil {
        return "", nil, err
    }
    if protocol != unixProtocol {
        return "", nil, fmt.Errorf("only support unix socket endpoint")
    }

    return addr, dial, nil
}

func dial(ctx context.Context, addr string) (net.Conn, error) {
    return (&net.Dialer{}).DialContext(ctx, unixProtocol, addr)
}

func parseEndpointWithFallbackProtocol(endpoint string, fallbackProtocol string) (protocol string, addr string, err error) {
    if protocol, addr, err = parseEndpoint(endpoint); err != nil && protocol == "" {
        fallbackEndpoint := fallbackProtocol + "://" + endpoint
        protocol, addr, err = parseEndpoint(fallbackEndpoint)
        if err == nil {
            log.Printf("Using %q as endpoint is deprecated, please consider using full url format %q.", endpoint, fallbackEndpoint)
        }
    }
    return
}

func parseEndpoint(endpoint string) (string, string, error) {
    u, err := url.Parse(endpoint)
    if err != nil {
        return "", "", err
    }
    switch u.Scheme {
    case "tcp":
        return "tcp", u.Host, nil
    case "unix":
        return "unix", u.Path, nil
    case "":
        return "", "", fmt.Errorf("using %q as endpoint is deprecated, please consider using full url format", endpoint)
    default:
        return u.Scheme, "", fmt.Errorf("protocol %q not supported", u.Scheme)
    }
}

func main() {

    runtimeClient, runtimeConn, err := getRuntimeClient()
    if err != nil {
        fmt.Println(err)
    }
    defer closeConnection(runtimeConn)

    request := &pb.ListContainersRequest{}
    listContainers, err := runtimeClient.ListContainers(context.Background(), request)
    if err != nil {
        fmt.Println(err)
    }

    for _, container := range listContainers.Containers {

        id := substr(container.GetId(), 12)
        var request = &pb.ContainerStatusRequest{ContainerId: id}
        info, err := runtimeClient.ContainerStatus(context.Background(), request)
        if err != nil {
            fmt.Println(err)
        }
        fmt.Println(info.Status.LogPath)
        var req = &pb.ContainerStatsRequest{ContainerId: id}
        stat, _ := runtimeClient.ContainerStats(context.Background(), req)
        if err != nil {
            fmt.Println(err)
        }
        if container.State.String() != "CONTAINER_RUNNING" {
            fmt.Printf("container %s it's not RUNNING, there are no log files and mount points\n", container.Metadata.Name)
            continue
        }
        fmt.Println(stat.Stats.WritableLayer.FsId.Mountpoint)
    }
}

func getRuntimeClient() (pb.RuntimeServiceClient, *grpc.ClientConn, error) {
    // Set up a connection to the server.
    conn, err := getConnection(RuntimeEndpoint)
    if err != nil {
        return nil, nil, errors.Wrap(err, "connect")
    }
    runtimeClient := pb.NewRuntimeServiceClient(conn)
    return runtimeClient, conn, nil
}

func closeConnection(conn *grpc.ClientConn) error {
    if conn == nil {
        return nil
    }
    return conn.Close()
}

func substr(s string, l int) string {
    if len(s) <= l {
        return s
    }
    ss, sl, rl, rs := "", 0, 0, []rune(s)
    for _, r := range rs {
        rint := int(r)
        if rint < 128 {
            rl = 1
        } else {
            rl = 2
        }
        if sl+rl > l {
            break
        }
        sl += rl
        ss += string(r)
    }
    return ss
}


func isExist(fileName string) bool{
    _,err := os.Stat(fileName)
    if err!=nil{
        if os.IsExist(err){
            return true
        }
        return false
    }
    return true
}