mirror of
https://github.com/siderolabs/talos.git
synced 2025-08-18 21:21:10 +02:00
573cce8d18
This PR introduces APId. This service replaces the frontend functionality previously provided by OSD. The main driver for this is two fold: 1. Create a single purpose application to expose the talos api 2. Make use of code generation to DRY api changes Signed-off-by: Brad Beam <brad.beam@talos-systems.com>
1027 lines
33 KiB
Go
1027 lines
33 KiB
Go
// Code generated by protoc-gen-go. DO NOT EDIT.
|
|
// source: api.proto
|
|
|
|
package api
|
|
|
|
import (
|
|
context "context"
|
|
fmt "fmt"
|
|
io "io"
|
|
math "math"
|
|
sync "sync"
|
|
|
|
proto "github.com/golang/protobuf/proto"
|
|
empty "github.com/golang/protobuf/ptypes/empty"
|
|
go_multierror "github.com/hashicorp/go-multierror"
|
|
grpc "google.golang.org/grpc"
|
|
credentials "google.golang.org/grpc/credentials"
|
|
metadata "google.golang.org/grpc/metadata"
|
|
|
|
common "github.com/talos-systems/talos/api/common"
|
|
machine "github.com/talos-systems/talos/api/machine"
|
|
os "github.com/talos-systems/talos/api/os"
|
|
constants "github.com/talos-systems/talos/pkg/constants"
|
|
tls "github.com/talos-systems/talos/pkg/grpc/tls"
|
|
)
|
|
|
|
// Reference imports to suppress errors if they are not otherwise used.
|
|
var _ = proto.Marshal
|
|
var _ = fmt.Errorf
|
|
var _ = math.Inf
|
|
|
|
// This is a compile-time assertion to ensure that this generated file
|
|
// is compatible with the proto package it is being compiled against.
|
|
// A compilation error at this line likely means your copy of the
|
|
// proto package needs to be updated.
|
|
const _ = proto.ProtoPackageIsVersion3 // please upgrade the proto package
|
|
|
|
// ContainersRequest from public import os/os.proto
|
|
type ContainersRequest = os.ContainersRequest
|
|
|
|
// Container from public import os/os.proto
|
|
type Container = os.Container
|
|
|
|
// ContainerResponse from public import os/os.proto
|
|
type ContainerResponse = os.ContainerResponse
|
|
|
|
// ContainersReply from public import os/os.proto
|
|
type ContainersReply = os.ContainersReply
|
|
|
|
// Data from public import os/os.proto
|
|
type Data = os.Data
|
|
|
|
// DataResponse from public import os/os.proto
|
|
type DataResponse = os.DataResponse
|
|
|
|
// DataReply from public import os/os.proto
|
|
type DataReply = os.DataReply
|
|
|
|
// LogsRequest from public import os/os.proto
|
|
type LogsRequest = os.LogsRequest
|
|
|
|
// ProcessesRequest from public import os/os.proto
|
|
type ProcessesRequest = os.ProcessesRequest
|
|
|
|
// ProcessesReply from public import os/os.proto
|
|
type ProcessesReply = os.ProcessesReply
|
|
|
|
// ProcessResponse from public import os/os.proto
|
|
type ProcessResponse = os.ProcessResponse
|
|
|
|
// Process from public import os/os.proto
|
|
type Process = os.Process
|
|
|
|
// RestartRequest from public import os/os.proto
|
|
type RestartRequest = os.RestartRequest
|
|
|
|
// RestartResponse from public import os/os.proto
|
|
type RestartResponse = os.RestartResponse
|
|
|
|
// RestartReply from public import os/os.proto
|
|
type RestartReply = os.RestartReply
|
|
|
|
// StatsRequest from public import os/os.proto
|
|
type StatsRequest = os.StatsRequest
|
|
|
|
// StatsResponse from public import os/os.proto
|
|
type StatsResponse = os.StatsResponse
|
|
|
|
// StatsReply from public import os/os.proto
|
|
type StatsReply = os.StatsReply
|
|
|
|
// Stat from public import os/os.proto
|
|
type Stat = os.Stat
|
|
|
|
// ContainerDriver from public import os/os.proto
|
|
type ContainerDriver = os.ContainerDriver
|
|
|
|
var ContainerDriver_name = os.ContainerDriver_name
|
|
var ContainerDriver_value = os.ContainerDriver_value
|
|
|
|
const ContainerDriver_CONTAINERD = ContainerDriver(os.ContainerDriver_CONTAINERD)
|
|
const ContainerDriver_CRI = ContainerDriver(os.ContainerDriver_CRI)
|
|
|
|
// RebootResponse from public import machine/machine.proto
|
|
type RebootResponse = machine.RebootResponse
|
|
|
|
// RebootReply from public import machine/machine.proto
|
|
type RebootReply = machine.RebootReply
|
|
|
|
// ResetResponse from public import machine/machine.proto
|
|
type ResetResponse = machine.ResetResponse
|
|
|
|
// ResetReply from public import machine/machine.proto
|
|
type ResetReply = machine.ResetReply
|
|
|
|
// ShutdownResponse from public import machine/machine.proto
|
|
type ShutdownResponse = machine.ShutdownResponse
|
|
|
|
// ShutdownReply from public import machine/machine.proto
|
|
type ShutdownReply = machine.ShutdownReply
|
|
|
|
// UpgradeRequest from public import machine/machine.proto
|
|
type UpgradeRequest = machine.UpgradeRequest
|
|
|
|
// UpgradeResponse from public import machine/machine.proto
|
|
type UpgradeResponse = machine.UpgradeResponse
|
|
|
|
// UpgradeReply from public import machine/machine.proto
|
|
type UpgradeReply = machine.UpgradeReply
|
|
|
|
// ServiceListResponse from public import machine/machine.proto
|
|
type ServiceListResponse = machine.ServiceListResponse
|
|
|
|
// ServiceListReply from public import machine/machine.proto
|
|
type ServiceListReply = machine.ServiceListReply
|
|
|
|
// ServiceInfo from public import machine/machine.proto
|
|
type ServiceInfo = machine.ServiceInfo
|
|
|
|
// ServiceEvents from public import machine/machine.proto
|
|
type ServiceEvents = machine.ServiceEvents
|
|
|
|
// ServiceEvent from public import machine/machine.proto
|
|
type ServiceEvent = machine.ServiceEvent
|
|
|
|
// ServiceHealth from public import machine/machine.proto
|
|
type ServiceHealth = machine.ServiceHealth
|
|
|
|
// ServiceStartRequest from public import machine/machine.proto
|
|
type ServiceStartRequest = machine.ServiceStartRequest
|
|
|
|
// ServiceStartResponse from public import machine/machine.proto
|
|
type ServiceStartResponse = machine.ServiceStartResponse
|
|
|
|
// ServiceStartReply from public import machine/machine.proto
|
|
type ServiceStartReply = machine.ServiceStartReply
|
|
|
|
// ServiceStopRequest from public import machine/machine.proto
|
|
type ServiceStopRequest = machine.ServiceStopRequest
|
|
|
|
// ServiceStopResponse from public import machine/machine.proto
|
|
type ServiceStopResponse = machine.ServiceStopResponse
|
|
|
|
// ServiceStopReply from public import machine/machine.proto
|
|
type ServiceStopReply = machine.ServiceStopReply
|
|
|
|
// ServiceRestartRequest from public import machine/machine.proto
|
|
type ServiceRestartRequest = machine.ServiceRestartRequest
|
|
|
|
// ServiceRestartResponse from public import machine/machine.proto
|
|
type ServiceRestartResponse = machine.ServiceRestartResponse
|
|
|
|
// ServiceRestartReply from public import machine/machine.proto
|
|
type ServiceRestartReply = machine.ServiceRestartReply
|
|
|
|
// StartRequest from public import machine/machine.proto
|
|
type StartRequest = machine.StartRequest
|
|
|
|
// StartReply from public import machine/machine.proto
|
|
type StartReply = machine.StartReply
|
|
|
|
// StopRequest from public import machine/machine.proto
|
|
type StopRequest = machine.StopRequest
|
|
|
|
// StopReply from public import machine/machine.proto
|
|
type StopReply = machine.StopReply
|
|
|
|
// StreamingData from public import machine/machine.proto
|
|
type StreamingData = machine.StreamingData
|
|
|
|
// CopyOutRequest from public import machine/machine.proto
|
|
type CopyOutRequest = machine.CopyOutRequest
|
|
|
|
// LSRequest from public import machine/machine.proto
|
|
type LSRequest = machine.LSRequest
|
|
|
|
// FileInfo from public import machine/machine.proto
|
|
type FileInfo = machine.FileInfo
|
|
|
|
// MountsResponse from public import machine/machine.proto
|
|
type MountsResponse = machine.MountsResponse
|
|
|
|
// MountsReply from public import machine/machine.proto
|
|
type MountsReply = machine.MountsReply
|
|
|
|
// MountStat from public import machine/machine.proto
|
|
type MountStat = machine.MountStat
|
|
|
|
// VersionResponse from public import machine/machine.proto
|
|
type VersionResponse = machine.VersionResponse
|
|
|
|
// VersionReply from public import machine/machine.proto
|
|
type VersionReply = machine.VersionReply
|
|
|
|
// VersionInfo from public import machine/machine.proto
|
|
type VersionInfo = machine.VersionInfo
|
|
|
|
// NodeMetadata from public import common/common.proto
|
|
type NodeMetadata = common.NodeMetadata
|
|
|
|
// Empty from public import google/protobuf/empty.proto
|
|
type Empty = empty.Empty
|
|
|
|
func init() { proto.RegisterFile("api.proto", fileDescriptor_00212fb1f9d3bf1c) }
|
|
|
|
var fileDescriptor_00212fb1f9d3bf1c = []byte{
|
|
// 140 bytes of a gzipped FileDescriptorProto
|
|
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x2c, 0x8b, 0xb1, 0x0a, 0xc2, 0x30,
|
|
0x14, 0x45, 0xd5, 0x82, 0xa0, 0x6e, 0x8a, 0x4b, 0xdd, 0xc4, 0xd5, 0xbe, 0xc1, 0x3f, 0xf0, 0x0b,
|
|
0x32, 0xbb, 0xa5, 0x21, 0xa6, 0x81, 0xbe, 0xde, 0x87, 0xef, 0x75, 0xe8, 0xdf, 0x0b, 0x4d, 0xa6,
|
|
0xc3, 0x39, 0x97, 0x7b, 0x3c, 0x78, 0xc9, 0x9d, 0xfc, 0x60, 0x38, 0x37, 0x5e, 0x72, 0x7b, 0x82,
|
|
0x12, 0xb4, 0x94, 0xf6, 0xca, 0x3e, 0x0c, 0x79, 0x8a, 0x54, 0x59, 0xf3, 0x25, 0x80, 0x19, 0x13,
|
|
0x15, 0xd4, 0x78, 0x4b, 0x40, 0x1a, 0x23, 0xad, 0xd6, 0xcf, 0x5f, 0x8a, 0x2c, 0xb6, 0x94, 0xf1,
|
|
0xfd, 0xf8, 0xdc, 0x53, 0xb6, 0x61, 0xee, 0xbb, 0x00, 0x26, 0xf3, 0x23, 0xf4, 0xa9, 0x8b, 0x5a,
|
|
0x64, 0x2d, 0x46, 0x5e, 0xb2, 0xdb, 0xb8, 0xad, 0xdb, 0xb9, 0xa6, 0xdf, 0xaf, 0xa7, 0xd7, 0x3f,
|
|
0x00, 0x00, 0xff, 0xff, 0x96, 0x64, 0x43, 0x69, 0x9c, 0x00, 0x00, 0x00,
|
|
}
|
|
|
|
type ApiProxy struct {
|
|
Provider tls.CertificateProvider
|
|
}
|
|
|
|
func NewApiProxy(provider tls.CertificateProvider) *ApiProxy {
|
|
return &ApiProxy{
|
|
Provider: provider,
|
|
}
|
|
}
|
|
|
|
func (p *ApiProxy) UnaryInterceptor() grpc.UnaryServerInterceptor {
|
|
return func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
|
|
md, _ := metadata.FromIncomingContext(ctx)
|
|
if _, ok := md["proxyfrom"]; ok {
|
|
return handler(ctx, req)
|
|
}
|
|
ca, err := p.Provider.GetCA()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
certs, err := p.Provider.GetCertificate(nil)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
tlsConfig, err := tls.New(
|
|
tls.WithClientAuthType(tls.Mutual),
|
|
tls.WithCACertPEM(ca),
|
|
tls.WithKeypair(*certs),
|
|
)
|
|
return p.Proxy(ctx, info.FullMethod, credentials.NewTLS(tlsConfig), req)
|
|
}
|
|
}
|
|
|
|
func (p *ApiProxy) Proxy(ctx context.Context, method string, creds credentials.TransportCredentials, in interface{}, opts ...grpc.CallOption) (proto.Message, error) {
|
|
var (
|
|
err error
|
|
errors *go_multierror.Error
|
|
msgs []proto.Message
|
|
ok bool
|
|
response proto.Message
|
|
targets []string
|
|
)
|
|
md, _ := metadata.FromIncomingContext(ctx)
|
|
// default to target node specified in config or on cli
|
|
if targets, ok = md["targets"]; !ok {
|
|
targets = md[":authority"]
|
|
}
|
|
proxyMd := metadata.New(make(map[string]string))
|
|
proxyMd.Set("proxyfrom", md[":authority"]...)
|
|
|
|
switch method {
|
|
case "/os.OS/Containers":
|
|
// Initialize target clients
|
|
clients, err := createOSClient(targets, creds, proxyMd)
|
|
if err != nil {
|
|
break
|
|
}
|
|
resp := &os.ContainersReply{}
|
|
msgs, err = proxyOSRunner(clients, in, proxyContainers)
|
|
for _, msg := range msgs {
|
|
resp.Response = append(resp.Response, msg.(*os.ContainersReply).Response[0])
|
|
}
|
|
response = resp
|
|
case "/os.OS/Dmesg":
|
|
// Initialize target clients
|
|
clients, err := createOSClient(targets, creds, proxyMd)
|
|
if err != nil {
|
|
break
|
|
}
|
|
resp := &os.DataReply{}
|
|
msgs, err = proxyOSRunner(clients, in, proxyDmesg)
|
|
for _, msg := range msgs {
|
|
resp.Response = append(resp.Response, msg.(*os.DataReply).Response[0])
|
|
}
|
|
response = resp
|
|
case "/os.OS/Kubeconfig":
|
|
// Initialize target clients
|
|
clients, err := createOSClient(targets, creds, proxyMd)
|
|
if err != nil {
|
|
break
|
|
}
|
|
resp := &os.DataReply{}
|
|
msgs, err = proxyOSRunner(clients, in, proxyKubeconfig)
|
|
for _, msg := range msgs {
|
|
resp.Response = append(resp.Response, msg.(*os.DataReply).Response[0])
|
|
}
|
|
response = resp
|
|
case "/os.OS/Processes":
|
|
// Initialize target clients
|
|
clients, err := createOSClient(targets, creds, proxyMd)
|
|
if err != nil {
|
|
break
|
|
}
|
|
resp := &os.ProcessesReply{}
|
|
msgs, err = proxyOSRunner(clients, in, proxyProcesses)
|
|
for _, msg := range msgs {
|
|
resp.Response = append(resp.Response, msg.(*os.ProcessesReply).Response[0])
|
|
}
|
|
response = resp
|
|
case "/os.OS/Restart":
|
|
// Initialize target clients
|
|
clients, err := createOSClient(targets, creds, proxyMd)
|
|
if err != nil {
|
|
break
|
|
}
|
|
resp := &os.RestartReply{}
|
|
msgs, err = proxyOSRunner(clients, in, proxyRestart)
|
|
for _, msg := range msgs {
|
|
resp.Response = append(resp.Response, msg.(*os.RestartReply).Response[0])
|
|
}
|
|
response = resp
|
|
case "/os.OS/Stats":
|
|
// Initialize target clients
|
|
clients, err := createOSClient(targets, creds, proxyMd)
|
|
if err != nil {
|
|
break
|
|
}
|
|
resp := &os.StatsReply{}
|
|
msgs, err = proxyOSRunner(clients, in, proxyStats)
|
|
for _, msg := range msgs {
|
|
resp.Response = append(resp.Response, msg.(*os.StatsReply).Response[0])
|
|
}
|
|
response = resp
|
|
case "/machine.Machine/Mounts":
|
|
// Initialize target clients
|
|
clients, err := createMachineClient(targets, creds, proxyMd)
|
|
if err != nil {
|
|
break
|
|
}
|
|
resp := &machine.MountsReply{}
|
|
msgs, err = proxyMachineRunner(clients, in, proxyMounts)
|
|
for _, msg := range msgs {
|
|
resp.Response = append(resp.Response, msg.(*machine.MountsReply).Response[0])
|
|
}
|
|
response = resp
|
|
case "/machine.Machine/Reboot":
|
|
// Initialize target clients
|
|
clients, err := createMachineClient(targets, creds, proxyMd)
|
|
if err != nil {
|
|
break
|
|
}
|
|
resp := &machine.RebootReply{}
|
|
msgs, err = proxyMachineRunner(clients, in, proxyReboot)
|
|
for _, msg := range msgs {
|
|
resp.Response = append(resp.Response, msg.(*machine.RebootReply).Response[0])
|
|
}
|
|
response = resp
|
|
case "/machine.Machine/Reset":
|
|
// Initialize target clients
|
|
clients, err := createMachineClient(targets, creds, proxyMd)
|
|
if err != nil {
|
|
break
|
|
}
|
|
resp := &machine.ResetReply{}
|
|
msgs, err = proxyMachineRunner(clients, in, proxyReset)
|
|
for _, msg := range msgs {
|
|
resp.Response = append(resp.Response, msg.(*machine.ResetReply).Response[0])
|
|
}
|
|
response = resp
|
|
case "/machine.Machine/Shutdown":
|
|
// Initialize target clients
|
|
clients, err := createMachineClient(targets, creds, proxyMd)
|
|
if err != nil {
|
|
break
|
|
}
|
|
resp := &machine.ShutdownReply{}
|
|
msgs, err = proxyMachineRunner(clients, in, proxyShutdown)
|
|
for _, msg := range msgs {
|
|
resp.Response = append(resp.Response, msg.(*machine.ShutdownReply).Response[0])
|
|
}
|
|
response = resp
|
|
case "/machine.Machine/Upgrade":
|
|
// Initialize target clients
|
|
clients, err := createMachineClient(targets, creds, proxyMd)
|
|
if err != nil {
|
|
break
|
|
}
|
|
resp := &machine.UpgradeReply{}
|
|
msgs, err = proxyMachineRunner(clients, in, proxyUpgrade)
|
|
for _, msg := range msgs {
|
|
resp.Response = append(resp.Response, msg.(*machine.UpgradeReply).Response[0])
|
|
}
|
|
response = resp
|
|
case "/machine.Machine/ServiceList":
|
|
// Initialize target clients
|
|
clients, err := createMachineClient(targets, creds, proxyMd)
|
|
if err != nil {
|
|
break
|
|
}
|
|
resp := &machine.ServiceListReply{}
|
|
msgs, err = proxyMachineRunner(clients, in, proxyServiceList)
|
|
for _, msg := range msgs {
|
|
resp.Response = append(resp.Response, msg.(*machine.ServiceListReply).Response[0])
|
|
}
|
|
response = resp
|
|
case "/machine.Machine/ServiceStart":
|
|
// Initialize target clients
|
|
clients, err := createMachineClient(targets, creds, proxyMd)
|
|
if err != nil {
|
|
break
|
|
}
|
|
resp := &machine.ServiceStartReply{}
|
|
msgs, err = proxyMachineRunner(clients, in, proxyServiceStart)
|
|
for _, msg := range msgs {
|
|
resp.Response = append(resp.Response, msg.(*machine.ServiceStartReply).Response[0])
|
|
}
|
|
response = resp
|
|
case "/machine.Machine/ServiceStop":
|
|
// Initialize target clients
|
|
clients, err := createMachineClient(targets, creds, proxyMd)
|
|
if err != nil {
|
|
break
|
|
}
|
|
resp := &machine.ServiceStopReply{}
|
|
msgs, err = proxyMachineRunner(clients, in, proxyServiceStop)
|
|
for _, msg := range msgs {
|
|
resp.Response = append(resp.Response, msg.(*machine.ServiceStopReply).Response[0])
|
|
}
|
|
response = resp
|
|
case "/machine.Machine/ServiceRestart":
|
|
// Initialize target clients
|
|
clients, err := createMachineClient(targets, creds, proxyMd)
|
|
if err != nil {
|
|
break
|
|
}
|
|
resp := &machine.ServiceRestartReply{}
|
|
msgs, err = proxyMachineRunner(clients, in, proxyServiceRestart)
|
|
for _, msg := range msgs {
|
|
resp.Response = append(resp.Response, msg.(*machine.ServiceRestartReply).Response[0])
|
|
}
|
|
response = resp
|
|
case "/machine.Machine/Version":
|
|
// Initialize target clients
|
|
clients, err := createMachineClient(targets, creds, proxyMd)
|
|
if err != nil {
|
|
break
|
|
}
|
|
resp := &machine.VersionReply{}
|
|
msgs, err = proxyMachineRunner(clients, in, proxyVersion)
|
|
for _, msg := range msgs {
|
|
resp.Response = append(resp.Response, msg.(*machine.VersionReply).Response[0])
|
|
}
|
|
response = resp
|
|
|
|
}
|
|
|
|
if err != nil {
|
|
errors = go_multierror.Append(errors, err)
|
|
}
|
|
return response, errors.ErrorOrNil()
|
|
}
|
|
|
|
func copyClientServer(msg interface{}, client grpc.ClientStream, srv grpc.ServerStream) error {
|
|
for {
|
|
err := client.RecvMsg(msg)
|
|
if err == io.EOF {
|
|
break
|
|
}
|
|
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
err = srv.SendMsg(msg)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
type runnerOSFn func(*proxyOSClient, interface{}, *sync.WaitGroup, chan proto.Message, chan error)
|
|
|
|
func proxyOSRunner(clients []*proxyOSClient, in interface{}, runner runnerOSFn) ([]proto.Message, error) {
|
|
var (
|
|
errors *go_multierror.Error
|
|
wg sync.WaitGroup
|
|
)
|
|
respCh := make(chan proto.Message, len(clients))
|
|
errCh := make(chan error, len(clients))
|
|
wg.Add(len(clients))
|
|
for _, client := range clients {
|
|
go runner(client, in, &wg, respCh, errCh)
|
|
}
|
|
wg.Wait()
|
|
close(respCh)
|
|
close(errCh)
|
|
|
|
var response []proto.Message
|
|
for resp := range respCh {
|
|
response = append(response, resp)
|
|
}
|
|
for err := range errCh {
|
|
errors = go_multierror.Append(errors, err)
|
|
}
|
|
return response, errors.ErrorOrNil()
|
|
}
|
|
|
|
type proxyOSClient struct {
|
|
Conn os.OSClient
|
|
Context context.Context
|
|
Target string
|
|
DialOpts []grpc.DialOption
|
|
}
|
|
|
|
func proxyContainers(client *proxyOSClient, in interface{}, wg *sync.WaitGroup, respCh chan proto.Message, errCh chan error) {
|
|
defer wg.Done()
|
|
resp, err := client.Conn.Containers(client.Context, in.(*os.ContainersRequest))
|
|
if err != nil {
|
|
errCh <- err
|
|
return
|
|
}
|
|
resp.Response[0].Metadata = &NodeMetadata{Hostname: client.Target}
|
|
respCh <- resp
|
|
}
|
|
|
|
func proxyDmesg(client *proxyOSClient, in interface{}, wg *sync.WaitGroup, respCh chan proto.Message, errCh chan error) {
|
|
defer wg.Done()
|
|
resp, err := client.Conn.Dmesg(client.Context, in.(*empty.Empty))
|
|
if err != nil {
|
|
errCh <- err
|
|
return
|
|
}
|
|
resp.Response[0].Metadata = &NodeMetadata{Hostname: client.Target}
|
|
respCh <- resp
|
|
}
|
|
|
|
func proxyKubeconfig(client *proxyOSClient, in interface{}, wg *sync.WaitGroup, respCh chan proto.Message, errCh chan error) {
|
|
defer wg.Done()
|
|
resp, err := client.Conn.Kubeconfig(client.Context, in.(*empty.Empty))
|
|
if err != nil {
|
|
errCh <- err
|
|
return
|
|
}
|
|
resp.Response[0].Metadata = &NodeMetadata{Hostname: client.Target}
|
|
respCh <- resp
|
|
}
|
|
|
|
func proxyProcesses(client *proxyOSClient, in interface{}, wg *sync.WaitGroup, respCh chan proto.Message, errCh chan error) {
|
|
defer wg.Done()
|
|
resp, err := client.Conn.Processes(client.Context, in.(*empty.Empty))
|
|
if err != nil {
|
|
errCh <- err
|
|
return
|
|
}
|
|
resp.Response[0].Metadata = &NodeMetadata{Hostname: client.Target}
|
|
respCh <- resp
|
|
}
|
|
|
|
func proxyRestart(client *proxyOSClient, in interface{}, wg *sync.WaitGroup, respCh chan proto.Message, errCh chan error) {
|
|
defer wg.Done()
|
|
resp, err := client.Conn.Restart(client.Context, in.(*os.RestartRequest))
|
|
if err != nil {
|
|
errCh <- err
|
|
return
|
|
}
|
|
resp.Response[0].Metadata = &NodeMetadata{Hostname: client.Target}
|
|
respCh <- resp
|
|
}
|
|
|
|
func proxyStats(client *proxyOSClient, in interface{}, wg *sync.WaitGroup, respCh chan proto.Message, errCh chan error) {
|
|
defer wg.Done()
|
|
resp, err := client.Conn.Stats(client.Context, in.(*os.StatsRequest))
|
|
if err != nil {
|
|
errCh <- err
|
|
return
|
|
}
|
|
resp.Response[0].Metadata = &NodeMetadata{Hostname: client.Target}
|
|
respCh <- resp
|
|
}
|
|
|
|
type runnerMachineFn func(*proxyMachineClient, interface{}, *sync.WaitGroup, chan proto.Message, chan error)
|
|
|
|
func proxyMachineRunner(clients []*proxyMachineClient, in interface{}, runner runnerMachineFn) ([]proto.Message, error) {
|
|
var (
|
|
errors *go_multierror.Error
|
|
wg sync.WaitGroup
|
|
)
|
|
respCh := make(chan proto.Message, len(clients))
|
|
errCh := make(chan error, len(clients))
|
|
wg.Add(len(clients))
|
|
for _, client := range clients {
|
|
go runner(client, in, &wg, respCh, errCh)
|
|
}
|
|
wg.Wait()
|
|
close(respCh)
|
|
close(errCh)
|
|
|
|
var response []proto.Message
|
|
for resp := range respCh {
|
|
response = append(response, resp)
|
|
}
|
|
for err := range errCh {
|
|
errors = go_multierror.Append(errors, err)
|
|
}
|
|
return response, errors.ErrorOrNil()
|
|
}
|
|
|
|
type proxyMachineClient struct {
|
|
Conn machine.MachineClient
|
|
Context context.Context
|
|
Target string
|
|
DialOpts []grpc.DialOption
|
|
}
|
|
|
|
func proxyMounts(client *proxyMachineClient, in interface{}, wg *sync.WaitGroup, respCh chan proto.Message, errCh chan error) {
|
|
defer wg.Done()
|
|
resp, err := client.Conn.Mounts(client.Context, in.(*empty.Empty))
|
|
if err != nil {
|
|
errCh <- err
|
|
return
|
|
}
|
|
resp.Response[0].Metadata = &NodeMetadata{Hostname: client.Target}
|
|
respCh <- resp
|
|
}
|
|
|
|
func proxyReboot(client *proxyMachineClient, in interface{}, wg *sync.WaitGroup, respCh chan proto.Message, errCh chan error) {
|
|
defer wg.Done()
|
|
resp, err := client.Conn.Reboot(client.Context, in.(*empty.Empty))
|
|
if err != nil {
|
|
errCh <- err
|
|
return
|
|
}
|
|
resp.Response[0].Metadata = &NodeMetadata{Hostname: client.Target}
|
|
respCh <- resp
|
|
}
|
|
|
|
func proxyReset(client *proxyMachineClient, in interface{}, wg *sync.WaitGroup, respCh chan proto.Message, errCh chan error) {
|
|
defer wg.Done()
|
|
resp, err := client.Conn.Reset(client.Context, in.(*empty.Empty))
|
|
if err != nil {
|
|
errCh <- err
|
|
return
|
|
}
|
|
resp.Response[0].Metadata = &NodeMetadata{Hostname: client.Target}
|
|
respCh <- resp
|
|
}
|
|
|
|
func proxyShutdown(client *proxyMachineClient, in interface{}, wg *sync.WaitGroup, respCh chan proto.Message, errCh chan error) {
|
|
defer wg.Done()
|
|
resp, err := client.Conn.Shutdown(client.Context, in.(*empty.Empty))
|
|
if err != nil {
|
|
errCh <- err
|
|
return
|
|
}
|
|
resp.Response[0].Metadata = &NodeMetadata{Hostname: client.Target}
|
|
respCh <- resp
|
|
}
|
|
|
|
func proxyUpgrade(client *proxyMachineClient, in interface{}, wg *sync.WaitGroup, respCh chan proto.Message, errCh chan error) {
|
|
defer wg.Done()
|
|
resp, err := client.Conn.Upgrade(client.Context, in.(*machine.UpgradeRequest))
|
|
if err != nil {
|
|
errCh <- err
|
|
return
|
|
}
|
|
resp.Response[0].Metadata = &NodeMetadata{Hostname: client.Target}
|
|
respCh <- resp
|
|
}
|
|
|
|
func proxyServiceList(client *proxyMachineClient, in interface{}, wg *sync.WaitGroup, respCh chan proto.Message, errCh chan error) {
|
|
defer wg.Done()
|
|
resp, err := client.Conn.ServiceList(client.Context, in.(*empty.Empty))
|
|
if err != nil {
|
|
errCh <- err
|
|
return
|
|
}
|
|
resp.Response[0].Metadata = &NodeMetadata{Hostname: client.Target}
|
|
respCh <- resp
|
|
}
|
|
|
|
func proxyServiceStart(client *proxyMachineClient, in interface{}, wg *sync.WaitGroup, respCh chan proto.Message, errCh chan error) {
|
|
defer wg.Done()
|
|
resp, err := client.Conn.ServiceStart(client.Context, in.(*machine.ServiceStartRequest))
|
|
if err != nil {
|
|
errCh <- err
|
|
return
|
|
}
|
|
resp.Response[0].Metadata = &NodeMetadata{Hostname: client.Target}
|
|
respCh <- resp
|
|
}
|
|
|
|
func proxyServiceStop(client *proxyMachineClient, in interface{}, wg *sync.WaitGroup, respCh chan proto.Message, errCh chan error) {
|
|
defer wg.Done()
|
|
resp, err := client.Conn.ServiceStop(client.Context, in.(*machine.ServiceStopRequest))
|
|
if err != nil {
|
|
errCh <- err
|
|
return
|
|
}
|
|
resp.Response[0].Metadata = &NodeMetadata{Hostname: client.Target}
|
|
respCh <- resp
|
|
}
|
|
|
|
func proxyServiceRestart(client *proxyMachineClient, in interface{}, wg *sync.WaitGroup, respCh chan proto.Message, errCh chan error) {
|
|
defer wg.Done()
|
|
resp, err := client.Conn.ServiceRestart(client.Context, in.(*machine.ServiceRestartRequest))
|
|
if err != nil {
|
|
errCh <- err
|
|
return
|
|
}
|
|
resp.Response[0].Metadata = &NodeMetadata{Hostname: client.Target}
|
|
respCh <- resp
|
|
}
|
|
|
|
func proxyVersion(client *proxyMachineClient, in interface{}, wg *sync.WaitGroup, respCh chan proto.Message, errCh chan error) {
|
|
defer wg.Done()
|
|
resp, err := client.Conn.Version(client.Context, in.(*empty.Empty))
|
|
if err != nil {
|
|
errCh <- err
|
|
return
|
|
}
|
|
resp.Response[0].Metadata = &NodeMetadata{Hostname: client.Target}
|
|
respCh <- resp
|
|
}
|
|
|
|
func createOSClient(targets []string, creds credentials.TransportCredentials, proxyMd metadata.MD) ([]*proxyOSClient, error) {
|
|
var errors *go_multierror.Error
|
|
clients := make([]*proxyOSClient, 0, len(targets))
|
|
for _, target := range targets {
|
|
c := &proxyOSClient{
|
|
// TODO change the context to be more useful ( ex cancelable )
|
|
Context: metadata.NewOutgoingContext(context.Background(), proxyMd),
|
|
Target: target,
|
|
}
|
|
// TODO: i think we potentially leak a client here,
|
|
// we should close the request // cancel the context if it errors
|
|
// Explicitly set OSD port
|
|
conn, err := grpc.Dial(fmt.Sprintf("%s:%d", target, 50000), grpc.WithTransportCredentials(creds))
|
|
if err != nil {
|
|
// TODO: probably worth wrapping err to add some context about the target
|
|
errors = go_multierror.Append(errors, err)
|
|
continue
|
|
}
|
|
c.Conn = os.NewOSClient(conn)
|
|
clients = append(clients, c)
|
|
}
|
|
return clients, errors.ErrorOrNil()
|
|
}
|
|
|
|
func createMachineClient(targets []string, creds credentials.TransportCredentials, proxyMd metadata.MD) ([]*proxyMachineClient, error) {
|
|
var errors *go_multierror.Error
|
|
clients := make([]*proxyMachineClient, 0, len(targets))
|
|
for _, target := range targets {
|
|
c := &proxyMachineClient{
|
|
// TODO change the context to be more useful ( ex cancelable )
|
|
Context: metadata.NewOutgoingContext(context.Background(), proxyMd),
|
|
Target: target,
|
|
}
|
|
// TODO: i think we potentially leak a client here,
|
|
// we should close the request // cancel the context if it errors
|
|
// Explicitly set OSD port
|
|
conn, err := grpc.Dial(fmt.Sprintf("%s:%d", target, 50000), grpc.WithTransportCredentials(creds))
|
|
if err != nil {
|
|
// TODO: probably worth wrapping err to add some context about the target
|
|
errors = go_multierror.Append(errors, err)
|
|
continue
|
|
}
|
|
c.Conn = machine.NewMachineClient(conn)
|
|
clients = append(clients, c)
|
|
}
|
|
return clients, errors.ErrorOrNil()
|
|
}
|
|
|
|
type Registrator struct {
|
|
os.OSClient
|
|
machine.MachineClient
|
|
}
|
|
|
|
func (r *Registrator) Register(s *grpc.Server) {
|
|
os.RegisterOSServer(s, r)
|
|
machine.RegisterMachineServer(s, r)
|
|
}
|
|
|
|
func (r *Registrator) Containers(ctx context.Context, in *os.ContainersRequest) (*os.ContainersReply, error) {
|
|
return r.OSClient.Containers(ctx, in)
|
|
}
|
|
|
|
func (r *Registrator) Dmesg(ctx context.Context, in *empty.Empty) (*os.DataReply, error) {
|
|
return r.OSClient.Dmesg(ctx, in)
|
|
}
|
|
|
|
func (r *Registrator) Kubeconfig(ctx context.Context, in *empty.Empty) (*os.DataReply, error) {
|
|
return r.OSClient.Kubeconfig(ctx, in)
|
|
}
|
|
|
|
func (r *Registrator) Logs(in *os.LogsRequest, srv os.OS_LogsServer) error {
|
|
client, err := r.OSClient.Logs(srv.Context(), in)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
var msg os.Data
|
|
return copyClientServer(&msg, client, srv)
|
|
}
|
|
|
|
func (r *Registrator) Processes(ctx context.Context, in *empty.Empty) (*os.ProcessesReply, error) {
|
|
return r.OSClient.Processes(ctx, in)
|
|
}
|
|
|
|
func (r *Registrator) Restart(ctx context.Context, in *os.RestartRequest) (*os.RestartReply, error) {
|
|
return r.OSClient.Restart(ctx, in)
|
|
}
|
|
|
|
func (r *Registrator) Stats(ctx context.Context, in *os.StatsRequest) (*os.StatsReply, error) {
|
|
return r.OSClient.Stats(ctx, in)
|
|
}
|
|
|
|
func (r *Registrator) CopyOut(in *machine.CopyOutRequest, srv machine.Machine_CopyOutServer) error {
|
|
client, err := r.MachineClient.CopyOut(srv.Context(), in)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
var msg machine.StreamingData
|
|
return copyClientServer(&msg, client, srv)
|
|
}
|
|
|
|
func (r *Registrator) Mounts(ctx context.Context, in *empty.Empty) (*machine.MountsReply, error) {
|
|
return r.MachineClient.Mounts(ctx, in)
|
|
}
|
|
|
|
func (r *Registrator) LS(in *machine.LSRequest, srv machine.Machine_LSServer) error {
|
|
client, err := r.MachineClient.LS(srv.Context(), in)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
var msg machine.FileInfo
|
|
return copyClientServer(&msg, client, srv)
|
|
}
|
|
|
|
func (r *Registrator) Reboot(ctx context.Context, in *empty.Empty) (*machine.RebootReply, error) {
|
|
return r.MachineClient.Reboot(ctx, in)
|
|
}
|
|
|
|
func (r *Registrator) Reset(ctx context.Context, in *empty.Empty) (*machine.ResetReply, error) {
|
|
return r.MachineClient.Reset(ctx, in)
|
|
}
|
|
|
|
func (r *Registrator) Shutdown(ctx context.Context, in *empty.Empty) (*machine.ShutdownReply, error) {
|
|
return r.MachineClient.Shutdown(ctx, in)
|
|
}
|
|
|
|
func (r *Registrator) Upgrade(ctx context.Context, in *machine.UpgradeRequest) (*machine.UpgradeReply, error) {
|
|
return r.MachineClient.Upgrade(ctx, in)
|
|
}
|
|
|
|
func (r *Registrator) ServiceList(ctx context.Context, in *empty.Empty) (*machine.ServiceListReply, error) {
|
|
return r.MachineClient.ServiceList(ctx, in)
|
|
}
|
|
|
|
func (r *Registrator) ServiceStart(ctx context.Context, in *machine.ServiceStartRequest) (*machine.ServiceStartReply, error) {
|
|
return r.MachineClient.ServiceStart(ctx, in)
|
|
}
|
|
|
|
func (r *Registrator) ServiceStop(ctx context.Context, in *machine.ServiceStopRequest) (*machine.ServiceStopReply, error) {
|
|
return r.MachineClient.ServiceStop(ctx, in)
|
|
}
|
|
|
|
func (r *Registrator) ServiceRestart(ctx context.Context, in *machine.ServiceRestartRequest) (*machine.ServiceRestartReply, error) {
|
|
return r.MachineClient.ServiceRestart(ctx, in)
|
|
}
|
|
|
|
func (r *Registrator) Start(ctx context.Context, in *machine.StartRequest) (*machine.StartReply, error) {
|
|
return r.MachineClient.Start(ctx, in)
|
|
}
|
|
|
|
func (r *Registrator) Stop(ctx context.Context, in *machine.StopRequest) (*machine.StopReply, error) {
|
|
return r.MachineClient.Stop(ctx, in)
|
|
}
|
|
|
|
func (r *Registrator) Version(ctx context.Context, in *empty.Empty) (*machine.VersionReply, error) {
|
|
return r.MachineClient.Version(ctx, in)
|
|
}
|
|
|
|
type LocalOSClient struct {
|
|
os.OSClient
|
|
}
|
|
|
|
func NewLocalOSClient() (os.OSClient, error) {
|
|
conn, err := grpc.Dial("unix:"+constants.OSSocketPath,
|
|
grpc.WithInsecure(),
|
|
)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return &LocalOSClient{
|
|
OSClient: os.NewOSClient(conn),
|
|
}, nil
|
|
}
|
|
|
|
func (c *LocalOSClient) Containers(ctx context.Context, in *os.ContainersRequest, opts ...grpc.CallOption) (*os.ContainersReply, error) {
|
|
return c.OSClient.Containers(ctx, in, opts...)
|
|
}
|
|
|
|
func (c *LocalOSClient) Dmesg(ctx context.Context, in *empty.Empty, opts ...grpc.CallOption) (*os.DataReply, error) {
|
|
return c.OSClient.Dmesg(ctx, in, opts...)
|
|
}
|
|
|
|
func (c *LocalOSClient) Kubeconfig(ctx context.Context, in *empty.Empty, opts ...grpc.CallOption) (*os.DataReply, error) {
|
|
return c.OSClient.Kubeconfig(ctx, in, opts...)
|
|
}
|
|
|
|
func (c *LocalOSClient) Logs(ctx context.Context, in *os.LogsRequest, opts ...grpc.CallOption) (os.OS_LogsClient, error) {
|
|
return c.OSClient.Logs(ctx, in, opts...)
|
|
}
|
|
|
|
func (c *LocalOSClient) Processes(ctx context.Context, in *empty.Empty, opts ...grpc.CallOption) (*os.ProcessesReply, error) {
|
|
return c.OSClient.Processes(ctx, in, opts...)
|
|
}
|
|
|
|
func (c *LocalOSClient) Restart(ctx context.Context, in *os.RestartRequest, opts ...grpc.CallOption) (*os.RestartReply, error) {
|
|
return c.OSClient.Restart(ctx, in, opts...)
|
|
}
|
|
|
|
func (c *LocalOSClient) Stats(ctx context.Context, in *os.StatsRequest, opts ...grpc.CallOption) (*os.StatsReply, error) {
|
|
return c.OSClient.Stats(ctx, in, opts...)
|
|
}
|
|
|
|
type LocalMachineClient struct {
|
|
machine.MachineClient
|
|
}
|
|
|
|
func NewLocalMachineClient() (machine.MachineClient, error) {
|
|
conn, err := grpc.Dial("unix:"+constants.MachineSocketPath,
|
|
grpc.WithInsecure(),
|
|
)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return &LocalMachineClient{
|
|
MachineClient: machine.NewMachineClient(conn),
|
|
}, nil
|
|
}
|
|
|
|
func (c *LocalMachineClient) CopyOut(ctx context.Context, in *machine.CopyOutRequest, opts ...grpc.CallOption) (machine.Machine_CopyOutClient, error) {
|
|
return c.MachineClient.CopyOut(ctx, in, opts...)
|
|
}
|
|
|
|
func (c *LocalMachineClient) Mounts(ctx context.Context, in *empty.Empty, opts ...grpc.CallOption) (*machine.MountsReply, error) {
|
|
return c.MachineClient.Mounts(ctx, in, opts...)
|
|
}
|
|
|
|
func (c *LocalMachineClient) LS(ctx context.Context, in *machine.LSRequest, opts ...grpc.CallOption) (machine.Machine_LSClient, error) {
|
|
return c.MachineClient.LS(ctx, in, opts...)
|
|
}
|
|
|
|
func (c *LocalMachineClient) Reboot(ctx context.Context, in *empty.Empty, opts ...grpc.CallOption) (*machine.RebootReply, error) {
|
|
return c.MachineClient.Reboot(ctx, in, opts...)
|
|
}
|
|
|
|
func (c *LocalMachineClient) Reset(ctx context.Context, in *empty.Empty, opts ...grpc.CallOption) (*machine.ResetReply, error) {
|
|
return c.MachineClient.Reset(ctx, in, opts...)
|
|
}
|
|
|
|
func (c *LocalMachineClient) Shutdown(ctx context.Context, in *empty.Empty, opts ...grpc.CallOption) (*machine.ShutdownReply, error) {
|
|
return c.MachineClient.Shutdown(ctx, in, opts...)
|
|
}
|
|
|
|
func (c *LocalMachineClient) Upgrade(ctx context.Context, in *machine.UpgradeRequest, opts ...grpc.CallOption) (*machine.UpgradeReply, error) {
|
|
return c.MachineClient.Upgrade(ctx, in, opts...)
|
|
}
|
|
|
|
func (c *LocalMachineClient) ServiceList(ctx context.Context, in *empty.Empty, opts ...grpc.CallOption) (*machine.ServiceListReply, error) {
|
|
return c.MachineClient.ServiceList(ctx, in, opts...)
|
|
}
|
|
|
|
func (c *LocalMachineClient) ServiceStart(ctx context.Context, in *machine.ServiceStartRequest, opts ...grpc.CallOption) (*machine.ServiceStartReply, error) {
|
|
return c.MachineClient.ServiceStart(ctx, in, opts...)
|
|
}
|
|
|
|
func (c *LocalMachineClient) ServiceStop(ctx context.Context, in *machine.ServiceStopRequest, opts ...grpc.CallOption) (*machine.ServiceStopReply, error) {
|
|
return c.MachineClient.ServiceStop(ctx, in, opts...)
|
|
}
|
|
|
|
func (c *LocalMachineClient) ServiceRestart(ctx context.Context, in *machine.ServiceRestartRequest, opts ...grpc.CallOption) (*machine.ServiceRestartReply, error) {
|
|
return c.MachineClient.ServiceRestart(ctx, in, opts...)
|
|
}
|
|
|
|
func (c *LocalMachineClient) Start(ctx context.Context, in *machine.StartRequest, opts ...grpc.CallOption) (*machine.StartReply, error) {
|
|
return c.MachineClient.Start(ctx, in, opts...)
|
|
}
|
|
|
|
func (c *LocalMachineClient) Stop(ctx context.Context, in *machine.StopRequest, opts ...grpc.CallOption) (*machine.StopReply, error) {
|
|
return c.MachineClient.Stop(ctx, in, opts...)
|
|
}
|
|
|
|
func (c *LocalMachineClient) Version(ctx context.Context, in *empty.Empty, opts ...grpc.CallOption) (*machine.VersionReply, error) {
|
|
return c.MachineClient.Version(ctx, in, opts...)
|
|
}
|