mirrors/talos
1
0
Fork 0
mirror of https://github.com/siderolabs/talos.git synced 2025-10-04 20:21:18 +02:00
Code Issues Packages Projects Releases Wiki Activity

refactor: merge osd into machined

Browse Source 
This merges `osd` API into `machined`. API was copied from `osd` into
`machined`, and `osd` API was deprecated.

For backwards compatibility, `machined` still implements `osd` API, so
older Talos API clients can still talk to the node without changes.

Docs were updated. No functional changes.

Signed-off-by: Andrey Smirnov <smirnov.andrey@gmail.com>
This commit is contained in:
Andrey Smirnov 2020-07-13 20:56:47 +03:00 committed by talos-bot
parent 19343e5f1a
commit cbb7ca8390
37 changed files with 3093 additions and 3083 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.conform.yaml
View File

@ -28,7 +28,6 @@ policies:
- networkd
- timed
- talosctl
- osd
- trustd
- routerd
- talosctl

23
Dockerfile
View File

@ -42,8 +42,6 @@ COPY ./api/common/common.proto /api/common/common.proto
RUN protoc -I/api --go_out=plugins=grpc,paths=source_relative:/api common/common.proto
COPY ./api/health/health.proto /api/health/health.proto
RUN protoc -I/api --go_out=plugins=grpc,paths=source_relative:/api health/health.proto
COPY ./api/os/os.proto /api/os/os.proto
RUN protoc -I/api --go_out=plugins=grpc,paths=source_relative:/api os/os.proto
COPY ./api/security/security.proto /api/security/security.proto
RUN protoc -I/api --go_out=plugins=grpc,paths=source_relative:/api security/security.proto
COPY ./api/machine/machine.proto /api/machine/machine.proto
@ -52,6 +50,8 @@ COPY ./api/time/time.proto /api/time/time.proto
RUN protoc -I/api --go_out=plugins=grpc,paths=source_relative:/api time/time.proto
COPY ./api/network/network.proto /api/network/network.proto
RUN protoc -I/api --go_out=plugins=grpc,paths=source_relative:/api network/network.proto
COPY ./api/os/os.proto /api/os/os.proto
RUN protoc -I/api --go_out=plugins=grpc,paths=source_relative:/api os/os.proto
# Gofumports generated files to adjust import order
RUN gofumports -w -local github.com/talos-systems/talos /api/
@ -141,24 +141,6 @@ WORKDIR /scratch
RUN printf "FROM scratch\nCOPY ./apid /apid\nENTRYPOINT [\"/apid\"]" > Dockerfile
RUN --security=insecure img build --tag ${USERNAME}/apid:${TAG} --output type=docker,dest=/apid.tar --no-console .
# The osd target builds the osd image.
FROM base AS osd-build
ARG SHA
ARG TAG
ARG VERSION_PKG="github.com/talos-systems/talos/pkg/version"
WORKDIR /src/internal/app/osd
RUN --mount=type=cache,target=/.cache/go-build go build -ldflags "-s -w -X ${VERSION_PKG}.Name=Server -X ${VERSION_PKG}.SHA=${SHA} -X ${VERSION_PKG}.Tag=${TAG}" -o /osd
RUN chmod +x /osd
FROM base AS osd-image
ARG TAG
ARG USERNAME
COPY --from=osd-build /osd /scratch/osd
WORKDIR /scratch
RUN printf "FROM scratch\nCOPY ./osd /osd\nENTRYPOINT [\"/osd\"]" > Dockerfile
RUN --security=insecure img build --tag ${USERNAME}/osd:${TAG} --output type=docker,dest=/osd.tar --no-console .
# The trustd target builds the trustd image.
FROM base AS trustd-build
@ -320,7 +302,6 @@ COPY --from=machined /machined /rootfs/sbin/init
COPY --from=apid-image /apid.tar /rootfs/usr/images/
COPY --from=bootkube-image /bootkube.tar /rootfs/usr/images/
COPY --from=timed-image /timed.tar /rootfs/usr/images/
COPY --from=osd-image /osd.tar /rootfs/usr/images/
COPY --from=trustd-image /trustd.tar /rootfs/usr/images/
COPY --from=networkd-image /networkd.tar /rootfs/usr/images/
COPY --from=routerd-image /routerd.tar /rootfs/usr/images/

2561
api/machine/machine.pb.go
View File

File diff suppressed because it is too large Load Diff

161
api/machine/machine.proto
View File

@ -15,14 +15,19 @@ import "common/common.proto";
// The machine service definition.
service MachineService {
rpc Bootstrap(BootstrapRequest) returns (BootstrapResponse);
rpc Containers(ContainersRequest) returns (ContainersResponse);
rpc Copy(CopyRequest) returns (stream common.Data);
rpc Dmesg(DmesgRequest) returns (stream common.Data);
rpc Events(EventsRequest) returns (stream Event);
rpc Kubeconfig(google.protobuf.Empty) returns (stream common.Data);
rpc List(ListRequest) returns (stream FileInfo);
rpc Logs(LogsRequest) returns (stream common.Data);
rpc Memory(google.protobuf.Empty) returns (MemoryResponse);
rpc Mounts(google.protobuf.Empty) returns (MountsResponse);
rpc Processes(google.protobuf.Empty) returns (ProcessesResponse);
rpc Read(ReadRequest) returns (stream common.Data);
rpc Reboot(google.protobuf.Empty) returns (RebootResponse);
rpc Restart(RestartRequest) returns (RestartResponse);
rpc Rollback(RollbackRequest) returns (RollbackResponse);
rpc Reset(ResetRequest) returns (ResetResponse);
rpc Recover(RecoverRequest) returns (RecoverResponse);
@ -31,6 +36,7 @@ service MachineService {
rpc ServiceStart(ServiceStartRequest) returns (ServiceStartResponse);
rpc ServiceStop(ServiceStopRequest) returns (ServiceStopResponse);
rpc Shutdown(google.protobuf.Empty) returns (ShutdownResponse);
rpc Stats(StatsRequest) returns (StatsResponse);
rpc Upgrade(UpgradeRequest) returns (UpgradeResponse);
rpc Version(google.protobuf.Empty) returns (VersionResponse);
}
@ -318,3 +324,158 @@ message RollbackRequest {}
message Rollback { common.Metadata metadata = 1; }
message RollbackResponse { repeated Rollback messages = 1; }
// rpc Containers
message ContainersRequest {
string namespace = 1;
// driver might be default "containerd" or "cri"
common.ContainerDriver driver = 2;
}
// The messages message containing the requested containers.
message ContainerInfo {
string namespace = 1;
string id = 2;
string image = 3;
uint32 pid = 4;
string status = 5;
string pod_id = 6;
string name = 7;
}
// The messages message containing the requested containers.
message Container {
common.Metadata metadata = 1;
repeated ContainerInfo containers = 2;
}
message ContainersResponse { repeated Container messages = 1; }
// dmesg
message DmesgRequest {
bool follow = 1;
bool tail = 2;
}
// rpc processes
message ProcessesRequest {}
message ProcessesResponse { repeated Process messages = 1; }
message Process {
common.Metadata metadata = 1;
repeated ProcessInfo processes = 2;
}
message ProcessInfo {
int32 pid = 1;
int32 ppid = 2;
string state = 3;
int32 threads = 4;
double cpu_time = 5;
uint64 virtual_memory = 6;
uint64 resident_memory = 7;
string command = 8;
string executable = 9;
string args = 10;
}
// rpc restart
// The request message containing the process to restart.
message RestartRequest {
string namespace = 1;
string id = 2;
// driver might be default "containerd" or "cri"
common.ContainerDriver driver = 3;
}
message Restart { common.Metadata metadata = 1; }
// The messages message containing the restart status.
message RestartResponse { repeated Restart messages = 1; }
// rpc stats
// The request message containing the containerd namespace.
message StatsRequest {
string namespace = 1;
// driver might be default "containerd" or "cri"
common.ContainerDriver driver = 2;
}
// The messages message containing the requested stats.
message Stats {
common.Metadata metadata = 1;
repeated Stat stats = 2;
}
message StatsResponse { repeated Stats messages = 1; }
// The messages message containing the requested stat.
message Stat {
string namespace = 1;
string id = 2;
uint64 memory_usage = 4;
uint64 cpu_usage = 5;
string pod_id = 6;
string name = 7;
}
message Memory {
common.Metadata metadata = 1;
MemInfo meminfo = 2;
}
message MemoryResponse { repeated Memory messages = 1; }
message MemInfo {
uint64 memtotal = 1;
uint64 memfree = 2;
uint64 memavailable = 3;
uint64 buffers = 4;
uint64 cached = 5;
uint64 swapcached = 6;
uint64 active = 7;
uint64 inactive = 8;
uint64 activeanon = 9;
uint64 inactiveanon = 10;
uint64 activefile = 11;
uint64 inactivefile = 12;
uint64 unevictable = 13;
uint64 mlocked = 14;
uint64 swaptotal = 15;
uint64 swapfree = 16;
uint64 dirty = 17;
uint64 writeback = 18;
uint64 anonpages = 19;
uint64 mapped = 20;
uint64 shmem = 21;
uint64 slab = 22;
uint64 sreclaimable = 23;
uint64 sunreclaim = 24;
uint64 kernelstack = 25;
uint64 pagetables = 26;
uint64 nfsunstable = 27;
uint64 bounce = 28;
uint64 writebacktmp = 29;
uint64 commitlimit = 30;
uint64 committedas = 31;
uint64 vmalloctotal = 32;
uint64 vmallocused = 33;
uint64 vmallocchunk = 34;
uint64 hardwarecorrupted = 35;
uint64 anonhugepages = 36;
uint64 shmemhugepages = 37;
uint64 shmempmdmapped = 38;
uint64 cmatotal = 39;
uint64 cmafree = 40;
uint64 hugepagestotal = 41;
uint64 hugepagesfree = 42;
uint64 hugepagesrsvd = 43;
uint64 hugepagessurp = 44;
uint64 hugepagesize = 45;
uint64 directmap4k = 46;
uint64 directmap2m = 47;
uint64 directmap1g = 48;
}

2183
api/os/os.pb.go
View File

File diff suppressed because it is too large Load Diff

172
api/os/os.proto
View File

@ -9,170 +9,18 @@ option java_package = "com.os.api";
import "google/protobuf/empty.proto";
import "common/common.proto";
import "machine/machine.proto";
// The OS service definition.
//
// OS Service also implements all the API of Init Service
// Deprecated: this API is deprecated and merged into Machine API.
service OSService {
rpc Containers(ContainersRequest) returns (ContainersResponse);
rpc Dmesg(DmesgRequest) returns (stream common.Data);
rpc Memory(google.protobuf.Empty) returns (MemoryResponse);
rpc Processes(google.protobuf.Empty) returns (ProcessesResponse);
rpc Restart(RestartRequest) returns (RestartResponse);
rpc Stats(StatsRequest) returns (StatsResponse);
}
// rpc Containers
message ContainersRequest {
string namespace = 1;
// driver might be default "containerd" or "cri"
common.ContainerDriver driver = 2;
}
// The messages message containing the requested containers.
message ContainerInfo {
string namespace = 1;
string id = 2;
string image = 3;
uint32 pid = 4;
string status = 5;
string pod_id = 6;
string name = 7;
}
// The messages message containing the requested containers.
message Container {
common.Metadata metadata = 1;
repeated ContainerInfo containers = 2;
}
message ContainersResponse { repeated Container messages = 1; }
// dmesg
message DmesgRequest {
bool follow = 1;
bool tail = 2;
}
// rpc processes
message ProcessesRequest {}
message ProcessesResponse { repeated Process messages = 1; }
message Process {
common.Metadata metadata = 1;
repeated ProcessInfo processes = 2;
}
message ProcessInfo {
int32 pid = 1;
int32 ppid = 2;
string state = 3;
int32 threads = 4;
double cpu_time = 5;
uint64 virtual_memory = 6;
uint64 resident_memory = 7;
string command = 8;
string executable = 9;
string args = 10;
}
// rpc restart
// The request message containing the process to restart.
message RestartRequest {
string namespace = 1;
string id = 2;
// driver might be default "containerd" or "cri"
common.ContainerDriver driver = 3;
}
message Restart { common.Metadata metadata = 1; }
// The messages message containing the restart status.
message RestartResponse { repeated Restart messages = 1; }
// rpc stats
// The request message containing the containerd namespace.
message StatsRequest {
string namespace = 1;
// driver might be default "containerd" or "cri"
common.ContainerDriver driver = 2;
}
// The messages message containing the requested stats.
message Stats {
common.Metadata metadata = 1;
repeated Stat stats = 2;
}
message StatsResponse { repeated Stats messages = 1; }
// The messages message containing the requested stat.
message Stat {
string namespace = 1;
string id = 2;
uint64 memory_usage = 4;
uint64 cpu_usage = 5;
string pod_id = 6;
string name = 7;
}
message Memory {
common.Metadata metadata = 1;
MemInfo meminfo = 2;
}
message MemoryResponse { repeated Memory messages = 1; }
message MemInfo {
uint64 memtotal = 1;
uint64 memfree = 2;
uint64 memavailable = 3;
uint64 buffers = 4;
uint64 cached = 5;
uint64 swapcached = 6;
uint64 active = 7;
uint64 inactive = 8;
uint64 activeanon = 9;
uint64 inactiveanon = 10;
uint64 activefile = 11;
uint64 inactivefile = 12;
uint64 unevictable = 13;
uint64 mlocked = 14;
uint64 swaptotal = 15;
uint64 swapfree = 16;
uint64 dirty = 17;
uint64 writeback = 18;
uint64 anonpages = 19;
uint64 mapped = 20;
uint64 shmem = 21;
uint64 slab = 22;
uint64 sreclaimable = 23;
uint64 sunreclaim = 24;
uint64 kernelstack = 25;
uint64 pagetables = 26;
uint64 nfsunstable = 27;
uint64 bounce = 28;
uint64 writebacktmp = 29;
uint64 commitlimit = 30;
uint64 committedas = 31;
uint64 vmalloctotal = 32;
uint64 vmallocused = 33;
uint64 vmallocchunk = 34;
uint64 hardwarecorrupted = 35;
uint64 anonhugepages = 36;
uint64 shmemhugepages = 37;
uint64 shmempmdmapped = 38;
uint64 cmatotal = 39;
uint64 cmafree = 40;
uint64 hugepagestotal = 41;
uint64 hugepagesfree = 42;
uint64 hugepagesrsvd = 43;
uint64 hugepagessurp = 44;
uint64 hugepagesize = 45;
uint64 directmap4k = 46;
uint64 directmap2m = 47;
uint64 directmap1g = 48;
option deprecated = true;
rpc Containers(machine.ContainersRequest) returns (machine.ContainersResponse);
rpc Dmesg(machine.DmesgRequest) returns (stream common.Data);
rpc Memory(google.protobuf.Empty) returns (machine.MemoryResponse);
rpc Processes(google.protobuf.Empty) returns (machine.ProcessesResponse);
rpc Restart(machine.RestartRequest) returns (machine.RestartResponse);
rpc Stats(machine.StatsRequest) returns (machine.StatsResponse);
}

4
cmd/talosctl/cmd/talos/containers.go
View File

@ -18,7 +18,7 @@ import (
"google.golang.org/grpc/peer"
"github.com/talos-systems/talos/api/common"
osapi "github.com/talos-systems/talos/api/os"
machineapi "github.com/talos-systems/talos/api/machine"
"github.com/talos-systems/talos/pkg/cli"
"github.com/talos-systems/talos/pkg/client"
"github.com/talos-systems/talos/pkg/constants"
@ -60,7 +60,7 @@ var containersCmd = &cobra.Command{
},
}
func containerRender(remotePeer *peer.Peer, resp *osapi.ContainersResponse) error {
func containerRender(remotePeer *peer.Peer, resp *machineapi.ContainersResponse) error {
w := tabwriter.NewWriter(os.Stdout, 0, 0, 3, ' ', 0)
fmt.Fprintln(w, "NODE\tNAMESPACE\tID\tIMAGE\tPID\tSTATUS")

6
cmd/talosctl/cmd/talos/memory.go
View File

@ -14,7 +14,7 @@ import (
"google.golang.org/grpc"
"google.golang.org/grpc/peer"
osapi "github.com/talos-systems/talos/api/os"
machineapi "github.com/talos-systems/talos/api/machine"
"github.com/talos-systems/talos/pkg/cli"
"github.com/talos-systems/talos/pkg/client"
)
@ -50,7 +50,7 @@ var memoryCmd = &cobra.Command{
},
}
func briefRender(remotePeer *peer.Peer, resp *osapi.MemoryResponse) error {
func briefRender(remotePeer *peer.Peer, resp *machineapi.MemoryResponse) error {
w := tabwriter.NewWriter(os.Stdout, 0, 0, 3, ' ', 0)
fmt.Fprintln(w, "NODE\tTOTAL\tUSED\tFREE\tSHARED\tBUFFERS\tCACHE\tAVAILABLE")
@ -79,7 +79,7 @@ func briefRender(remotePeer *peer.Peer, resp *osapi.MemoryResponse) error {
return w.Flush()
}
func verboseRender(remotePeer *peer.Peer, resp *osapi.MemoryResponse) error {
func verboseRender(remotePeer *peer.Peer, resp *machineapi.MemoryResponse) error {
defaultNode := client.AddrFromPeer(remotePeer)
// Dump as /proc/meminfo

14
cmd/talosctl/cmd/talos/processes.go
View File

@ -21,7 +21,7 @@ import (
"google.golang.org/grpc"
"google.golang.org/grpc/peer"
osapi "github.com/talos-systems/talos/api/os"
machineapi "github.com/talos-systems/talos/api/machine"
"github.com/talos-systems/talos/pkg/cli"
"github.com/talos-systems/talos/pkg/client"
)
@ -128,9 +128,9 @@ func processesUI(ctx context.Context, c *client.Client) {
}
}
type by func(p1, p2 *osapi.ProcessInfo) bool
type by func(p1, p2 *machineapi.ProcessInfo) bool
func (b by) sort(procs []*osapi.ProcessInfo) {
func (b by) sort(procs []*machineapi.ProcessInfo) {
ps := &procSorter{
procs: procs,
by: b, // The Sort method's receiver is the function (closure) that defines the sort order.
@ -139,8 +139,8 @@ func (b by) sort(procs []*osapi.ProcessInfo) {
}
type procSorter struct {
procs []*osapi.ProcessInfo
by func(p1, p2 *osapi.ProcessInfo) bool // Closure used in the Less method.
procs []*machineapi.ProcessInfo
by func(p1, p2 *machineapi.ProcessInfo) bool // Closure used in the Less method.
}
// Len is part of sort.Interface.
@ -159,12 +159,12 @@ func (s *procSorter) Less(i, j int) bool {
}
// Sort Methods.
var rss = func(p1, p2 *osapi.ProcessInfo) bool {
var rss = func(p1, p2 *machineapi.ProcessInfo) bool {
// Reverse sort ( Descending )
return p1.ResidentMemory > p2.ResidentMemory
}
var cpu = func(p1, p2 *osapi.ProcessInfo) bool {
var cpu = func(p1, p2 *machineapi.ProcessInfo) bool {
// Reverse sort ( Descending )
return p1.CpuTime > p2.CpuTime
}

4
cmd/talosctl/cmd/talos/stats.go
View File

@ -18,7 +18,7 @@ import (
"google.golang.org/grpc/peer"
"github.com/talos-systems/talos/api/common"
osapi "github.com/talos-systems/talos/api/os"
machineapi "github.com/talos-systems/talos/api/machine"
"github.com/talos-systems/talos/pkg/cli"
"github.com/talos-systems/talos/pkg/client"
"github.com/talos-systems/talos/pkg/constants"
@ -59,7 +59,7 @@ var statsCmd = &cobra.Command{
},
}
func statsRender(remotePeer *peer.Peer, resp *osapi.StatsResponse) error {
func statsRender(remotePeer *peer.Peer, resp *machineapi.StatsResponse) error {
w := tabwriter.NewWriter(os.Stdout, 0, 0, 3, ' ', 0)
fmt.Fprintln(w, "NODE\tNAMESPACE\tID\tMEMORY(MB)\tCPU")

4
docs/website/content/v0.6.en.json
View File

@ -166,10 +166,6 @@
"title": "timed",
"path": "v0.6/en/components/timed"
},
{
"title": "osd",
"path": "v0.6/en/components/osd"
},
{
"title": "trustd",
"path": "v0.6/en/components/trustd"

8
docs/website/content/v0.6/en/components/apid.md
View File

@ -17,7 +17,7 @@ If `--nodes` is not specified, the first endpoint will be used.
> Note: Typically there will be an `endpoint` already defined in the Talos config file.
> Optionally, `nodes` can be included here as well.
For example, if a user wants to interact with `osd`, a command like `talosctl -e cluster.talos.dev memory` may be used.
For example, if a user wants to interact with `machined`, a command like `talosctl -e cluster.talos.dev memory` may be used.
```bash
$ talosctl -e cluster.talos.dev memory
@ -25,7 +25,7 @@ NODE TOTAL USED FREE SHARED BUFFERS CACHE AVAILABLE
cluster.talos.dev 7938 1768 2390 145 53 3724 6571
```
In this case, `talosctl` is interacting with `apid` running on `cluster.talos.dev` and forwarding the request to the `osd` api.
In this case, `talosctl` is interacting with `apid` running on `cluster.talos.dev` and forwarding the request to the `machined` api.
If we wanted to extend our example to retrieve `memory` from another node in our cluster, we could use the command `talosctl -e cluster.talos.dev -n node02 memory`.
@ -35,7 +35,7 @@ NODE TOTAL USED FREE SHARED BUFFERS CACHE AVAILABLE
node02 7938 1768 2390 145 53 3724 6571
```
The `apid` instance on `cluster.talos.dev` receives the request and forwards it to `apid` running on `node02` which forwards the request to the `osd` api.
The `apid` instance on `cluster.talos.dev` receives the request and forwards it to `apid` running on `node02` which forwards the request to the `machined` api.
We can further extend our example to retrieve `memory` for all nodes in our cluster by appending additional `-n node` flags or using a comma separated list of nodes ( `-n node01,node02,node03` ):
@ -47,4 +47,4 @@ node02 257844 14408 190796 18138 49 52589 227492
node03 257844 1830 255186 125 49 777 254556
```
The `apid` instance on `cluster.talos.dev` receives the request and forwards is to `node01`, `node02`, and `node03` which then forwards the request to their local `osd` api.
The `apid` instance on `cluster.talos.dev` receives the request and forwards is to `node01`, `node02`, and `node03` which then forwards the request to their local `machined` api.

1
docs/website/content/v0.6/en/components/machined.md
View File

@ -16,6 +16,5 @@ This includes:
- [kubelet](https://kubernetes.io/docs/concepts/overview/components/)
- [networkd](networkd)
- [timed](timed)
- [osd](osd)
- [trustd](trustd)
- [udevd](udevd)

2
docs/website/content/v0.6/en/components/osctl.md
View File

@ -2,7 +2,7 @@
title: 'talosctl'
---
`talosctl` CLI is the client to the [osd](/components/osd) service running on every node.
`talosctl` CLI is the client to the [apid](/components/apid) service running on every node.
`talosctl` should provide enough functionality to be a replacement for typical interactive shell operations.
With it you can do things like:

20
docs/website/content/v0.6/en/components/osd.md
View File

@ -1,20 +0,0 @@
---
title: 'osd'
---
Talos is unique in that it has no concept of host-level access.
There is no ssh daemon.
There is no interactive console session.
There are no shells installed.
Only what is required to run Kubernetes.
Furthermore, there is no way to run any custom processes on the host level.
To make this work, we needed an out-of-band tool for managing the nodes.
In an ideal world, the system would be self-healing and we would never have to touch it.
But, in the real world, this does not happen.
We still need a way to handle operational scenarios that may arise.
The `osd` daemon provides a way to do just that.
Based on the Principle of Least Privilege, `osd` provides operational value for cluster administrators by providing an API for node management.
Interactions with `osd` are handled via [talosctl](/docs/components/talosctl) which communicates via gRPC.

1
docs/website/content/v0.6/en/components/overview.md
View File

@ -13,7 +13,6 @@ In this section we will discuss the various components of which Talos is compris
| [machined](machined) | Talos replacement for the traditional Linux init-process. Specially designed to run Kubernetes and does not allow starting arbitrary user services. |
| [networkd](networkd) | Handles all of the host level network configuration. Configuration is defined under the `networking` key |
| [timed](timed) | Handles the host time synchronization by acting as a NTP-client. |
| [osd](osd) | Because there's no concept of host-level access in Talos, this is the out-of-band management tool for the nodes. It provides access to node information and offers operational functions. |
| [kernel](kernel) | The Linux kernel included with Talos is configured according to the recommendations outlined in the [Kernel Self Protection Project](http://kernsec.org/wiki/index.php/Kernel_Self_Protection_Project). |
| [routerd](routerd) | Responsible for routing an incoming API request from `apid` to the appropriate backend (e.g. `osd`, `machined` and `timed`). |
| [trustd](trustd) | To run and operate a Kubernetes cluster a certain level of trust is required. Based on the concept of a 'Root of Trust', `trustd` is a simple daemon responsible for establishing trust within the system. |

2
docs/website/content/v0.6/en/configuration/overview.md
View File

@ -5,7 +5,7 @@ title: 'Configuration Overview'
In this section, we will step through the configuration of a Talos based Kubernetes cluster.
There are three major components we will configure:
- `osd` and `talosctl`
- `apid` and `talosctl`
- the master nodes
- the worker nodes

4
docs/website/content/v0.6/en/guides/cloud/azure.md
View File

@ -80,11 +80,11 @@ az network vnet create \
# Create network security group
az network nsg create -g $GROUP -n talos-sg
# Client -> OSD
# Client -> apid
az network nsg rule create \
-g $GROUP \
--nsg-name talos-sg \
-n osd \
-n apid \
--priority 1001 \
--destination-port-ranges 50000 \
--direction inbound

2
docs/website/content/v0.6/en/guides/getting-started/talosctl.md
View File

@ -104,7 +104,6 @@ NODE SERVICE STATE HEALTH LAST CHANGE LAST EVENT
192.168.2.44 machined-api Running ? 192h7m48s ago Service started as goroutine
192.168.2.44 networkd Running OK 192h7m11s ago Health check successful
192.168.2.44 ntpd Running ? 192h7m10s ago Started task ntpd (PID 4144) for container ntpd
192.168.2.44 osd Running OK 192h7m45s ago Health check successful
192.168.2.44 routerd Running OK 192h7m46s ago Started task routerd (PID 3907) for container routerd
192.168.2.44 system-containerd Running OK 192h7m48s ago Health check successful
192.168.2.44 trustd Running OK 192h7m45s ago Health check successful
@ -125,7 +124,6 @@ NODE NAMESPACE ID IMAGE PID STATUS
192.168.2.44 system apid talos/apid 4021 RUNNING
192.168.2.44 system networkd talos/networkd 3893 RUNNING
192.168.2.44 system ntpd talos/ntpd 4144 RUNNING
192.168.2.44 system osd talos/osd 4086 RUNNING
192.168.2.44 system routerd talos/routerd 3907 RUNNING
192.168.2.44 system trustd talos/trustd 4010 RUNNING
```

1
internal/app/apid/main.go
View File

@ -72,6 +72,7 @@ func main() {
// all existing streaming methods
for _, methodName := range []string{
"/machine.MachineService/Copy",
"/machine.MachineService/Dmesg",
"/machine.MachineService/Events",
"/machine.MachineService/Kubeconfig",
"/machine.MachineService/List",

19
internal/app/machined/internal/server/v1alpha1/osd_compat.go Normal file
View File

@ -0,0 +1,19 @@
// This Source Code Form is subject to the terms of the Mozilla Public
// License, v. 2.0. If a copy of the MPL was not distributed with this
// file, You can obtain one at http://mozilla.org/MPL/2.0/.
package runtime
import (
"github.com/talos-systems/talos/api/machine"
"github.com/talos-systems/talos/api/os"
)
type osdServer struct {
*Server
}
// Dmesg implements the osapi.OsServer interface.
func (s *osdServer) Dmesg(req *machine.DmesgRequest, srv os.OSService_DmesgServer) error {
return s.Server.Dmesg(req, machine.MachineService_DmesgServer(srv))
}

331
internal/app/machined/internal/server/v1alpha1/v1alpha1_server.go
View File

@ -17,6 +17,7 @@ import (
"os"
"path/filepath"
"strings"
"syscall"
"time"
"github.com/containerd/containerd"
@ -26,12 +27,14 @@ import (
criconstants "github.com/containerd/cri/pkg/constants"
"github.com/golang/protobuf/ptypes/empty"
"github.com/hashicorp/go-multierror"
"github.com/prometheus/procfs"
"github.com/rs/xid"
"golang.org/x/sys/unix"
"google.golang.org/grpc"
"github.com/talos-systems/talos/api/common"
"github.com/talos-systems/talos/api/machine"
osapi "github.com/talos-systems/talos/api/os"
"github.com/talos-systems/talos/internal/app/machined/pkg/runtime"
"github.com/talos-systems/talos/internal/app/machined/pkg/runtime/v1alpha1/bootloader/syslinux"
"github.com/talos-systems/talos/internal/app/machined/pkg/system"
@ -40,6 +43,7 @@ import (
"github.com/talos-systems/talos/internal/pkg/containers/cri"
"github.com/talos-systems/talos/internal/pkg/containers/image"
"github.com/talos-systems/talos/internal/pkg/etcd"
"github.com/talos-systems/talos/internal/pkg/kmsg"
"github.com/talos-systems/talos/internal/pkg/kubeconfig"
"github.com/talos-systems/talos/pkg/archiver"
"github.com/talos-systems/talos/pkg/chunker"
@ -63,6 +67,7 @@ func (s *Server) Register(obj *grpc.Server) {
s.server = obj
machine.RegisterMachineServiceServer(obj, s)
osapi.RegisterOSServiceServer(obj, &osdServer{Server: s}) //nolint: staticcheck
}
// Reboot implements the machine.MachineServer interface.
@ -462,7 +467,7 @@ func (s *Server) List(req *machine.ListRequest, obj machine.MachineService_ListS
return nil
}
// Mounts implements the machine.OSDServer interface.
// Mounts implements the machine.MachineServer interface.
func (s *Server) Mounts(ctx context.Context, in *empty.Empty) (reply *machine.MountsResponse, err error) {
file, err := os.Open("/proc/mounts")
if err != nil {
@ -553,7 +558,7 @@ func (s *Server) Version(ctx context.Context, in *empty.Empty) (reply *machine.V
}, nil
}
// Kubeconfig implements the osapi.OSDServer interface.
// Kubeconfig implements the machine.MachineServer interface.
func (s *Server) Kubeconfig(empty *empty.Empty, obj machine.MachineService_KubeconfigServer) error {
var b bytes.Buffer
@ -839,3 +844,325 @@ func pullAndValidateInstallerImage(ctx context.Context, reg runtime.Registries,
return nil
}
// Containers implements the machine.MachineServer interface.
func (s *Server) Containers(ctx context.Context, in *machine.ContainersRequest) (reply *machine.ContainersResponse, err error) {
inspector, err := getContainerInspector(ctx, in.Namespace, in.Driver)
if err != nil {
return nil, err
}
// nolint: errcheck
defer inspector.Close()
pods, err := inspector.Pods()
if err != nil {
// fatal error
if pods == nil {
return nil, err
}
// TODO: only some failed, need to handle it better via client
log.Println(err.Error())
}
containers := []*machine.ContainerInfo{}
for _, pod := range pods {
for _, container := range pod.Containers {
container := &machine.ContainerInfo{
Namespace: in.Namespace,
Id: container.Display,
PodId: pod.Name,
Name: container.Name,
Image: container.Image,
Pid: container.Pid,
Status: container.Status,
}
containers = append(containers, container)
}
}
reply = &machine.ContainersResponse{
Messages: []*machine.Container{
{
Containers: containers,
},
},
}
return reply, nil
}
// Stats implements the machine.MachineServer interface.
// nolint: gocyclo
func (s *Server) Stats(ctx context.Context, in *machine.StatsRequest) (reply *machine.StatsResponse, err error) {
inspector, err := getContainerInspector(ctx, in.Namespace, in.Driver)
if err != nil {
return nil, err
}
// nolint: errcheck
defer inspector.Close()
pods, err := inspector.Pods()
if err != nil {
// fatal error
if pods == nil {
return nil, err
}
// TODO: only some failed, need to handle it better via client
log.Println(err.Error())
}
stats := []*machine.Stat{}
for _, pod := range pods {
for _, container := range pod.Containers {
if container.Metrics == nil {
continue
}
stat := &machine.Stat{
Namespace: in.Namespace,
Id: container.Display,
PodId: pod.Name,
Name: container.Name,
MemoryUsage: container.Metrics.MemoryUsage,
CpuUsage: container.Metrics.CPUUsage,
}
stats = append(stats, stat)
}
}
reply = &machine.StatsResponse{
Messages: []*machine.Stats{
{
Stats: stats,
},
},
}
return reply, nil
}
// Restart implements the machine.MachineServer interface.
func (s *Server) Restart(ctx context.Context, in *machine.RestartRequest) (*machine.RestartResponse, error) {
inspector, err := getContainerInspector(ctx, in.Namespace, in.Driver)
if err != nil {
return nil, err
}
// nolint: errcheck
defer inspector.Close()
container, err := inspector.Container(in.Id)
if err != nil {
return nil, err
}
if container == nil {
return nil, fmt.Errorf("container %q not found", in.Id)
}
err = container.Kill(syscall.SIGTERM)
if err != nil {
return nil, err
}
return &machine.RestartResponse{
Messages: []*machine.Restart{
{},
},
}, nil
}
// Dmesg implements the machine.MachineServer interface.
//
//nolint: gocyclo
func (s *Server) Dmesg(req *machine.DmesgRequest, srv machine.MachineService_DmesgServer) error {
ctx := srv.Context()
var options []kmsg.Option
if req.Follow {
options = append(options, kmsg.Follow())
}
if req.Tail {
options = append(options, kmsg.FromTail())
}
reader, err := kmsg.NewReader(options...)
if err != nil {
return fmt.Errorf("error opening /dev/kmsg reader: %w", err)
}
defer reader.Close() //nolint: errcheck
ch := reader.Scan(ctx)
for {
select {
case <-ctx.Done():
if err = reader.Close(); err != nil {
return err
}
case packet, ok := <-ch:
if !ok {
return nil
}
if packet.Err != nil {
err = srv.Send(&common.Data{
Metadata: &common.Metadata{
Error: packet.Err.Error(),
},
})
} else {
msg := packet.Message
err = srv.Send(&common.Data{
Bytes: []byte(fmt.Sprintf("%s: %7s: [%s]: %s", msg.Facility, msg.Priority, msg.Timestamp.Format(time.RFC3339Nano), msg.Message)),
})
}
if err != nil {
return err
}
}
}
}
// Processes implements the machine.MachineServer interface.
func (s *Server) Processes(ctx context.Context, in *empty.Empty) (reply *machine.ProcessesResponse, err error) {
procs, err := procfs.AllProcs()
if err != nil {
return nil, err
}
processes := make([]*machine.ProcessInfo, 0, len(procs))
var (
command string
executable string
args []string
stats procfs.ProcStat
)
for _, proc := range procs {
command, err = proc.Comm()
if err != nil {
return nil, err
}
executable, err = proc.Executable()
if err != nil {
return nil, err
}
args, err = proc.CmdLine()
if err != nil {
return nil, err
}
stats, err = proc.Stat()
if err != nil {
return nil, err
}
p := &machine.ProcessInfo{
Pid: int32(proc.PID),
Ppid: int32(stats.PPID),
State: stats.State,
Threads: int32(stats.NumThreads),
CpuTime: stats.CPUTime(),
VirtualMemory: uint64(stats.VirtualMemory()),
ResidentMemory: uint64(stats.ResidentMemory()),
Command: command,
Executable: executable,
Args: strings.Join(args, " "),
}
processes = append(processes, p)
}
reply = &machine.ProcessesResponse{
Messages: []*machine.Process{
{
Processes: processes,
},
},
}
return reply, nil
}
// Memory implements the machine.MachineServer interface.
func (s *Server) Memory(ctx context.Context, in *empty.Empty) (reply *machine.MemoryResponse, err error) {
proc, err := procfs.NewDefaultFS()
if err != nil {
return nil, err
}
info, err := proc.Meminfo()
if err != nil {
return nil, err
}
meminfo := &machine.MemInfo{
Memtotal: info.MemTotal,
Memfree: info.MemFree,
Memavailable: info.MemAvailable,
Buffers: info.Buffers,
Cached: info.Cached,
Swapcached: info.SwapCached,
Active: info.Active,
Inactive: info.Inactive,
Activeanon: info.ActiveAnon,
Inactiveanon: info.InactiveAnon,
Activefile: info.ActiveFile,
Inactivefile: info.InactiveFile,
Unevictable: info.Unevictable,
Mlocked: info.Mlocked,
Swaptotal: info.SwapTotal,
Swapfree: info.SwapFree,
Dirty: info.Dirty,
Writeback: info.Writeback,
Anonpages: info.AnonPages,
Mapped: info.Mapped,
Shmem: info.Shmem,
Slab: info.Slab,
Sreclaimable: info.SReclaimable,
Sunreclaim: info.SUnreclaim,
Kernelstack: info.KernelStack,
Pagetables: info.PageTables,
Nfsunstable: info.NFSUnstable,
Bounce: info.Bounce,
Writebacktmp: info.WritebackTmp,
Commitlimit: info.CommitLimit,
Committedas: info.CommittedAS,
Vmalloctotal: info.VmallocTotal,
Vmallocused: info.VmallocUsed,
Vmallocchunk: info.VmallocChunk,
Hardwarecorrupted: info.HardwareCorrupted,
Anonhugepages: info.AnonHugePages,
Shmemhugepages: info.ShmemHugePages,
Shmempmdmapped: info.ShmemPmdMapped,
Cmatotal: info.CmaTotal,
Cmafree: info.CmaFree,
Hugepagestotal: info.HugePagesTotal,
Hugepagesfree: info.HugePagesFree,
Hugepagesrsvd: info.HugePagesRsvd,
Hugepagessurp: info.HugePagesSurp,
Hugepagesize: info.Hugepagesize,
Directmap4K: info.DirectMap4k,
Directmap2M: info.DirectMap2M,
Directmap1G: info.DirectMap1G,
}
reply = &machine.MemoryResponse{
Messages: []*machine.Memory{
{
Meminfo: meminfo,
},
},
}
return reply, err
}

1
internal/app/machined/pkg/runtime/v1alpha1/v1alpha1_sequencer_tasks.go
View File

@ -618,7 +618,6 @@ func StartAllServices(seq runtime.Sequence, data interface{}) (runtime.TaskExecu
&services.APID{},
&services.Routerd{},
&services.Networkd{},
&services.OSD{},
&services.CRI{},
&services.Kubelet{},
)

10
internal/app/machined/pkg/system/runner/containerd/containerd_test.go
View File

@ -369,9 +369,9 @@ func (suite *ContainerdSuite) TestStopSigKill() {
func (suite *ContainerdSuite) TestImportSuccess() {
reqs := []*containerdrunner.ImportRequest{
{
Path: "/usr/images/osd.tar",
Path: "/usr/images/timed.tar",
Options: []containerd.ImportOpt{
containerd.WithIndexName("testtalos/osd"),
containerd.WithIndexName("testtalos/timed"),
},
},
{
@ -386,7 +386,7 @@ func (suite *ContainerdSuite) TestImportSuccess() {
ctx := namespaces.WithNamespace(context.Background(), suite.containerdNamespace)
for _, imageName := range []string{"testtalos/osd", "testtalos/trustd"} {
for _, imageName := range []string{"testtalos/timed", "testtalos/trustd"} {
image, err := suite.client.ImageService().Get(ctx, imageName)
suite.Require().NoError(err)
suite.Require().Equal(imageName, image.Name)
@ -396,9 +396,9 @@ func (suite *ContainerdSuite) TestImportSuccess() {
func (suite *ContainerdSuite) TestImportFail() {
reqs := []*containerdrunner.ImportRequest{
{
Path: "/usr/images/osd.tar",
Path: "/usr/images/timed.tar",
Options: []containerd.ImportOpt{
containerd.WithIndexName("testtalos/osd2"),
containerd.WithIndexName("testtalos/timed2"),
},
},
{

143
internal/app/machined/pkg/system/services/osd.go
View File

@ -1,143 +0,0 @@
// This Source Code Form is subject to the terms of the Mozilla Public
// License, v. 2.0. If a copy of the MPL was not distributed with this
// file, You can obtain one at http://mozilla.org/MPL/2.0/.
// nolint: golint
package services
import (
"context"
"fmt"
"os"
"path"
"path/filepath"
"strings"
containerdapi "github.com/containerd/containerd"
"github.com/containerd/containerd/oci"
specs "github.com/opencontainers/runtime-spec/specs-go"
"github.com/syndtr/gocapability/capability"
"google.golang.org/grpc"
"github.com/talos-systems/talos/internal/app/machined/pkg/runtime"
"github.com/talos-systems/talos/internal/app/machined/pkg/system/events"
"github.com/talos-systems/talos/internal/app/machined/pkg/system/health"
"github.com/talos-systems/talos/internal/app/machined/pkg/system/runner"
"github.com/talos-systems/talos/internal/app/machined/pkg/system/runner/containerd"
"github.com/talos-systems/talos/internal/app/machined/pkg/system/runner/restart"
"github.com/talos-systems/talos/internal/pkg/conditions"
"github.com/talos-systems/talos/pkg/constants"
"github.com/talos-systems/talos/pkg/grpc/dialer"
)
// OSD implements the Service interface. It serves as the concrete type with
// the required methods.
type OSD struct{}
// ID implements the Service interface.
func (o *OSD) ID(r runtime.Runtime) string {
return "osd"
}
// PreFunc implements the Service interface.
func (o *OSD) PreFunc(ctx context.Context, r runtime.Runtime) error {
importer := containerd.NewImporter(constants.SystemContainerdNamespace, containerd.WithContainerdAddress(constants.SystemContainerdAddress))
return importer.Import(&containerd.ImportRequest{
Path: "/usr/images/osd.tar",
Options: []containerdapi.ImportOpt{
containerdapi.WithIndexName("talos/osd"),
},
})
}
// PostFunc implements the Service interface.
func (o *OSD) PostFunc(r runtime.Runtime, state events.ServiceState) (err error) {
return nil
}
// Condition implements the Service interface.
func (o *OSD) Condition(r runtime.Runtime) conditions.Condition {
return nil
}
// DependsOn implements the Service interface.
func (o *OSD) DependsOn(r runtime.Runtime) []string {
return []string{"containerd", "networkd"}
}
func (o *OSD) Runner(r runtime.Runtime) (runner.Runner, error) {
image := "talos/osd"
// Set the process arguments.
args := runner.Args{
ID: o.ID(r),
ProcessArgs: []string{
"/osd",
},
}
// Ensure socket dir exists
if err := os.MkdirAll(filepath.Dir(constants.OSSocketPath), 0750); err != nil {
return nil, err
}
// Set the mounts.
mounts := []specs.Mount{
{Type: "bind", Destination: "/etc/ssl", Source: "/etc/ssl", Options: []string{"bind", "ro"}},
{Type: "bind", Destination: "/tmp", Source: "/tmp", Options: []string{"rbind", "rshared", "rw"}},
{Type: "bind", Destination: constants.ConfigPath, Source: constants.ConfigPath, Options: []string{"rbind", "ro"}},
{Type: "bind", Destination: path.Dir(constants.ContainerdAddress), Source: path.Dir(constants.ContainerdAddress), Options: []string{"bind", "ro"}},
{Type: "bind", Destination: constants.SystemRunPath, Source: constants.SystemRunPath, Options: []string{"bind", "ro"}},
{Type: "bind", Destination: filepath.Dir(constants.OSSocketPath), Source: filepath.Dir(constants.OSSocketPath), Options: []string{"rbind", "rw"}},
}
env := []string{}
for key, val := range r.Config().Machine().Env() {
env = append(env, fmt.Sprintf("%s=%s", key, val))
}
return restart.New(containerd.NewRunner(
r.Config().Debug(),
&args,
runner.WithLoggingManager(r.Logging()),
runner.WithContainerdAddress(constants.SystemContainerdAddress),
runner.WithContainerImage(image),
runner.WithEnv(env),
runner.WithOCISpecOpts(
oci.WithCapabilities([]string{
strings.ToUpper("CAP_" + capability.CAP_SYS_PTRACE.String()),
strings.ToUpper("CAP_" + capability.CAP_DAC_READ_SEARCH.String()),
strings.ToUpper("CAP_" + capability.CAP_DAC_OVERRIDE.String()),
strings.ToUpper("CAP_" + capability.CAP_SYSLOG.String()),
}),
oci.WithHostNamespace(specs.PIDNamespace),
oci.WithMounts(mounts),
oci.WithLinuxDevice("/dev/kmsg", "r"),
),
),
restart.WithType(restart.Forever),
), nil
}
// HealthFunc implements the HealthcheckedService interface.
func (o *OSD) HealthFunc(runtime.Runtime) health.Check {
return func(ctx context.Context) error {
conn, err := grpc.DialContext(
ctx,
fmt.Sprintf("%s://%s", "unix", constants.OSSocketPath),
grpc.WithInsecure(),
grpc.WithContextDialer(dialer.DialUnix()),
)
if err != nil {
return err
}
return conn.Close()
}
}
// HealthSettings implements the HealthcheckedService interface.
func (o *OSD) HealthSettings(runtime.Runtime) *health.Settings {
return &health.DefaultSettings
}

18
internal/app/machined/pkg/system/services/osd_test.go
View File

@ -1,18 +0,0 @@
// This Source Code Form is subject to the terms of the Mozilla Public
// License, v. 2.0. If a copy of the MPL was not distributed with this
// file, You can obtain one at http://mozilla.org/MPL/2.0/.
package services_test
import (
"testing"
"github.com/stretchr/testify/assert"
"github.com/talos-systems/talos/internal/app/machined/pkg/system"
"github.com/talos-systems/talos/internal/app/machined/pkg/system/services"
)
func TestOSDInterfaces(t *testing.T) {
assert.Implements(t, (*system.HealthcheckedService)(nil), new(services.OSD))
}

2
internal/app/machined/pkg/system/system_test.go
View File

@ -22,7 +22,7 @@ func (suite *SystemServicesSuite) TestStartShutdown() {
system.Services(nil).LoadAndStart(
&MockService{name: "containerd"},
&MockService{name: "trustd", dependencies: []string{"containerd"}},
&MockService{name: "osd", dependencies: []string{"containerd", "trustd"}},
&MockService{name: "machined", dependencies: []string{"containerd", "trustd"}},
)
time.Sleep(10 * time.Millisecond)

379
internal/app/osd/internal/reg/reg.go
View File

@ -1,379 +0,0 @@
// This Source Code Form is subject to the terms of the Mozilla Public
// License, v. 2.0. If a copy of the MPL was not distributed with this
// file, You can obtain one at http://mozilla.org/MPL/2.0/.
package reg
import (
"context"
"errors"
"fmt"
"log"
"strings"
"syscall"
"time"
criconstants "github.com/containerd/cri/pkg/constants"
"github.com/golang/protobuf/ptypes/empty"
"github.com/prometheus/procfs"
"google.golang.org/grpc"
"github.com/talos-systems/talos/api/common"
osapi "github.com/talos-systems/talos/api/os"
"github.com/talos-systems/talos/internal/pkg/containers"
"github.com/talos-systems/talos/internal/pkg/containers/containerd"
"github.com/talos-systems/talos/internal/pkg/containers/cri"
"github.com/talos-systems/talos/internal/pkg/kmsg"
"github.com/talos-systems/talos/pkg/constants"
)
// Registrator is the concrete type that implements the factory.Registrator and
// osapi.OSDServer interfaces.
type Registrator struct{}
// Register implements the factory.Registrator interface.
func (r *Registrator) Register(s *grpc.Server) {
osapi.RegisterOSServiceServer(s, r)
}
// Containers implements the osapi.OSDServer interface.
func (r *Registrator) Containers(ctx context.Context, in *osapi.ContainersRequest) (reply *osapi.ContainersResponse, err error) {
inspector, err := getContainerInspector(ctx, in.Namespace, in.Driver)
if err != nil {
return nil, err
}
// nolint: errcheck
defer inspector.Close()
pods, err := inspector.Pods()
if err != nil {
// fatal error
if pods == nil {
return nil, err
}
// TODO: only some failed, need to handle it better via client
log.Println(err.Error())
}
containers := []*osapi.ContainerInfo{}
for _, pod := range pods {
for _, container := range pod.Containers {
container := &osapi.ContainerInfo{
Namespace: in.Namespace,
Id: container.Display,
PodId: pod.Name,
Name: container.Name,
Image: container.Image,
Pid: container.Pid,
Status: container.Status,
}
containers = append(containers, container)
}
}
reply = &osapi.ContainersResponse{
Messages: []*osapi.Container{
{
Containers: containers,
},
},
}
return reply, nil
}
// Stats implements the osapi.OSDServer interface.
// nolint: gocyclo
func (r *Registrator) Stats(ctx context.Context, in *osapi.StatsRequest) (reply *osapi.StatsResponse, err error) {
inspector, err := getContainerInspector(ctx, in.Namespace, in.Driver)
if err != nil {
return nil, err
}
// nolint: errcheck
defer inspector.Close()
pods, err := inspector.Pods()
if err != nil {
// fatal error
if pods == nil {
return nil, err
}
// TODO: only some failed, need to handle it better via client
log.Println(err.Error())
}
stats := []*osapi.Stat{}
for _, pod := range pods {
for _, container := range pod.Containers {
if container.Metrics == nil {
continue
}
stat := &osapi.Stat{
Namespace: in.Namespace,
Id: container.Display,
PodId: pod.Name,
Name: container.Name,
MemoryUsage: container.Metrics.MemoryUsage,
CpuUsage: container.Metrics.CPUUsage,
}
stats = append(stats, stat)
}
}
reply = &osapi.StatsResponse{
Messages: []*osapi.Stats{
{
Stats: stats,
},
},
}
return reply, nil
}
// Restart implements the osapi.OSDServer interface.
func (r *Registrator) Restart(ctx context.Context, in *osapi.RestartRequest) (*osapi.RestartResponse, error) {
inspector, err := getContainerInspector(ctx, in.Namespace, in.Driver)
if err != nil {
return nil, err
}
// nolint: errcheck
defer inspector.Close()
container, err := inspector.Container(in.Id)
if err != nil {
return nil, err
}
if container == nil {
return nil, fmt.Errorf("container %q not found", in.Id)
}
err = container.Kill(syscall.SIGTERM)
if err != nil {
return nil, err
}
return &osapi.RestartResponse{
Messages: []*osapi.Restart{
{},
},
}, nil
}
// Dmesg implements the osapi.OSDServer interface.
//
//nolint: gocyclo
func (r *Registrator) Dmesg(req *osapi.DmesgRequest, srv osapi.OSService_DmesgServer) error {
ctx := srv.Context()
var options []kmsg.Option
if req.Follow {
options = append(options, kmsg.Follow())
}
if req.Tail {
options = append(options, kmsg.FromTail())
}
reader, err := kmsg.NewReader(options...)
if err != nil {
return fmt.Errorf("error opening /dev/kmsg reader: %w", err)
}
defer reader.Close() //nolint: errcheck
ch := reader.Scan(ctx)
for {
select {
case <-ctx.Done():
if err = reader.Close(); err != nil {
return err
}
case packet, ok := <-ch:
if !ok {
return nil
}
if packet.Err != nil {
err = srv.Send(&common.Data{
Metadata: &common.Metadata{
Error: packet.Err.Error(),
},
})
} else {
msg := packet.Message
err = srv.Send(&common.Data{
Bytes: []byte(fmt.Sprintf("%s: %7s: [%s]: %s", msg.Facility, msg.Priority, msg.Timestamp.Format(time.RFC3339Nano), msg.Message)),
})
}
if err != nil {
return err
}
}
}
}
// Processes implements the osapi.OSDServer interface.
func (r *Registrator) Processes(ctx context.Context, in *empty.Empty) (reply *osapi.ProcessesResponse, err error) {
procs, err := procfs.AllProcs()
if err != nil {
return nil, err
}
processes := make([]*osapi.ProcessInfo, 0, len(procs))
var (
command string
executable string
args []string
stats procfs.ProcStat
)
for _, proc := range procs {
command, err = proc.Comm()
if err != nil {
return nil, err
}
executable, err = proc.Executable()
if err != nil {
return nil, err
}
args, err = proc.CmdLine()
if err != nil {
return nil, err
}
stats, err = proc.Stat()
if err != nil {
return nil, err
}
p := &osapi.ProcessInfo{
Pid: int32(proc.PID),
Ppid: int32(stats.PPID),
State: stats.State,
Threads: int32(stats.NumThreads),
CpuTime: stats.CPUTime(),
VirtualMemory: uint64(stats.VirtualMemory()),
ResidentMemory: uint64(stats.ResidentMemory()),
Command: command,
Executable: executable,
Args: strings.Join(args, " "),
}
processes = append(processes, p)
}
reply = &osapi.ProcessesResponse{
Messages: []*osapi.Process{
{
Processes: processes,
},
},
}
return reply, nil
}
func getContainerInspector(ctx context.Context, namespace string, driver common.ContainerDriver) (containers.Inspector, error) {
switch driver {
case common.ContainerDriver_CRI:
if namespace != criconstants.K8sContainerdNamespace {
return nil, errors.New("CRI inspector is supported only for K8s namespace")
}
return cri.NewInspector(ctx)
case common.ContainerDriver_CONTAINERD:
addr := constants.ContainerdAddress
if namespace == constants.SystemContainerdNamespace {
addr = constants.SystemContainerdAddress
}
return containerd.NewInspector(ctx, namespace, containerd.WithContainerdAddress(addr))
default:
return nil, fmt.Errorf("unsupported driver %q", driver)
}
}
// Memory implements the osdapi.OSDServer interface.
func (r *Registrator) Memory(ctx context.Context, in *empty.Empty) (reply *osapi.MemoryResponse, err error) {
proc, err := procfs.NewDefaultFS()
if err != nil {
return nil, err
}
info, err := proc.Meminfo()
if err != nil {
return nil, err
}
meminfo := &osapi.MemInfo{
Memtotal: info.MemTotal,
Memfree: info.MemFree,
Memavailable: info.MemAvailable,
Buffers: info.Buffers,
Cached: info.Cached,
Swapcached: info.SwapCached,
Active: info.Active,
Inactive: info.Inactive,
Activeanon: info.ActiveAnon,
Inactiveanon: info.InactiveAnon,
Activefile: info.ActiveFile,
Inactivefile: info.InactiveFile,
Unevictable: info.Unevictable,
Mlocked: info.Mlocked,
Swaptotal: info.SwapTotal,
Swapfree: info.SwapFree,
Dirty: info.Dirty,
Writeback: info.Writeback,
Anonpages: info.AnonPages,
Mapped: info.Mapped,
Shmem: info.Shmem,
Slab: info.Slab,
Sreclaimable: info.SReclaimable,
Sunreclaim: info.SUnreclaim,
Kernelstack: info.KernelStack,
Pagetables: info.PageTables,
Nfsunstable: info.NFSUnstable,
Bounce: info.Bounce,
Writebacktmp: info.WritebackTmp,
Commitlimit: info.CommitLimit,
Committedas: info.CommittedAS,
Vmalloctotal: info.VmallocTotal,
Vmallocused: info.VmallocUsed,
Vmallocchunk: info.VmallocChunk,
Hardwarecorrupted: info.HardwareCorrupted,
Anonhugepages: info.AnonHugePages,
Shmemhugepages: info.ShmemHugePages,
Shmempmdmapped: info.ShmemPmdMapped,
Cmatotal: info.CmaTotal,
Cmafree: info.CmaFree,
Hugepagestotal: info.HugePagesTotal,
Hugepagesfree: info.HugePagesFree,
Hugepagesrsvd: info.HugePagesRsvd,
Hugepagessurp: info.HugePagesSurp,
Hugepagesize: info.Hugepagesize,
Directmap4K: info.DirectMap4k,
Directmap2M: info.DirectMap2M,
Directmap1G: info.DirectMap1G,
}
reply = &osapi.MemoryResponse{
Messages: []*osapi.Memory{
{
Meminfo: meminfo,
},
},
}
return reply, err
}

5
internal/app/osd/internal/reg/reg_test.go
View File

@ -1,5 +0,0 @@
// This Source Code Form is subject to the terms of the Mozilla Public
// License, v. 2.0. If a copy of the MPL was not distributed with this
// file, You can obtain one at http://mozilla.org/MPL/2.0/.
package reg_test

32
internal/app/osd/main.go
View File

@ -1,32 +0,0 @@
// This Source Code Form is subject to the terms of the Mozilla Public
// License, v. 2.0. If a copy of the MPL was not distributed with this
// file, You can obtain one at http://mozilla.org/MPL/2.0/.
package main
import (
"log"
"github.com/talos-systems/talos/internal/app/osd/internal/reg"
"github.com/talos-systems/talos/pkg/constants"
"github.com/talos-systems/talos/pkg/grpc/factory"
"github.com/talos-systems/talos/pkg/startup"
)
func init() {
log.SetFlags(log.Lshortfile | log.Ldate | log.Lmicroseconds | log.Ltime)
}
func main() {
if err := startup.RandSeed(); err != nil {
log.Fatalf("failed to seed RNG: %v", err)
}
log.Fatalf("%+v", factory.ListenAndServe(
&reg.Registrator{},
factory.Network("unix"),
factory.SocketPath(constants.OSSocketPath),
factory.WithDefaultLog(),
),
)
}

5
internal/app/routerd/main.go
View File

@ -28,8 +28,9 @@ func main() {
router := director.NewRouter()
// TODO: this should be dynamic based on plugin registration
router.RegisterLocalBackend("os.OSService", backend.NewLocal("osd", constants.OSSocketPath))
router.RegisterLocalBackend("machine.MachineService", backend.NewLocal("machined", constants.MachineSocketPath))
machinedBackend := backend.NewLocal("machined", constants.MachineSocketPath)
router.RegisterLocalBackend("os.OSService", machinedBackend)
router.RegisterLocalBackend("machine.MachineService", machinedBackend)
router.RegisterLocalBackend("time.TimeService", backend.NewLocal("timed", constants.TimeSocketPath))
router.RegisterLocalBackend("network.NetworkService", backend.NewLocal("networkd", constants.NetworkSocketPath))

6
internal/integration/api/logs.go
View File

@ -165,7 +165,7 @@ func (suite *LogsSuite) TestTailStreaming0() {
func (suite *LogsSuite) testStreaming(tailLines int32) {
if tailLines >= 0 {
// invoke osd enough times to generate
// invoke machined enough times to generate
// some logs
for i := int32(0); i < tailLines; i++ {
_, err := suite.Client.Stats(suite.nodeCtx, constants.SystemContainerdNamespace, common.ContainerDriver_CONTAINERD)
@ -177,7 +177,7 @@ func (suite *LogsSuite) testStreaming(tailLines int32) {
suite.nodeCtx,
constants.SystemContainerdNamespace,
common.ContainerDriver_CONTAINERD,
"osd",
"machined",
true,
tailLines,
)
@ -229,7 +229,7 @@ DrainLoop:
suite.Assert().InDelta(tailLines, linesDrained, 1)
}
// invoke osd API
// invoke machined API
_, err = suite.Client.Stats(suite.nodeCtx, constants.SystemContainerdNamespace, common.ContainerDriver_CONTAINERD)
suite.Require().NoError(err)

2
internal/integration/cli/containers.go
View File

@ -26,7 +26,7 @@ func (suite *ContainersSuite) SuiteName() string {
func (suite *ContainersSuite) TestContainerd() {
suite.RunCLI([]string{"containers"},
base.StdoutShouldMatch(regexp.MustCompile(`IMAGE`)),
base.StdoutShouldMatch(regexp.MustCompile(`talos/osd`)),
base.StdoutShouldMatch(regexp.MustCompile(`talos/routerd`)),
)
suite.RunCLI([]string{"containers", "-k"},
base.StdoutShouldMatch(regexp.MustCompile(`kubelet`)),

6
internal/integration/cli/services.go
View File

@ -26,7 +26,7 @@ func (suite *ServicesSuite) SuiteName() string {
func (suite *ServicesSuite) TestList() {
suite.RunCLI([]string{"services"},
base.StdoutShouldMatch(regexp.MustCompile(`STATE`)),
base.StdoutShouldMatch(regexp.MustCompile(`osd`)),
base.StdoutShouldMatch(regexp.MustCompile(`routerd`)),
base.StdoutShouldMatch(regexp.MustCompile(`apid`)),
)
}
@ -39,9 +39,9 @@ func (suite *ServicesSuite) TestStatus() {
base.StdoutShouldMatch(regexp.MustCompile(`\[Running\]`)),
)
suite.RunCLI([]string{"service", "osd", "status"},
suite.RunCLI([]string{"service", "routerd", "status"},
base.StdoutShouldMatch(regexp.MustCompile(`STATE`)),
base.StdoutShouldMatch(regexp.MustCompile(`osd`)),
base.StdoutShouldMatch(regexp.MustCompile(`routerd`)),
base.StdoutShouldMatch(regexp.MustCompile(`\[Running\]`)),
)
}

2
internal/integration/cli/stats.go
View File

@ -26,7 +26,7 @@ func (suite *StatsSuite) SuiteName() string {
func (suite *StatsSuite) TestContainerd() {
suite.RunCLI([]string{"stats"},
base.StdoutShouldMatch(regexp.MustCompile(`CPU`)),
base.StdoutShouldMatch(regexp.MustCompile(`osd`)),
base.StdoutShouldMatch(regexp.MustCompile(`routerd`)),
)
suite.RunCLI([]string{"stats", "-k"},
base.StdoutShouldMatch(regexp.MustCompile(`CPU`)),

38
pkg/client/client.go
View File

@ -28,7 +28,6 @@ import (
"github.com/talos-systems/talos/api/common"
machineapi "github.com/talos-systems/talos/api/machine"
networkapi "github.com/talos-systems/talos/api/network"
osapi "github.com/talos-systems/talos/api/os"
timeapi "github.com/talos-systems/talos/api/time"
"github.com/talos-systems/talos/pkg/client/config"
"github.com/talos-systems/talos/pkg/constants"
@ -47,7 +46,6 @@ type Credentials struct {
type Client struct {
options *Options
conn *grpc.ClientConn
OSClient osapi.OSServiceClient
MachineClient machineapi.MachineServiceClient
TimeClient timeapi.TimeServiceClient
NetworkClient networkapi.NetworkServiceClient
@ -120,7 +118,6 @@ func New(ctx context.Context, opts ...OptionFunc) (c *Client, err error) {
return nil, fmt.Errorf("failed to create client connection: %w", err)
}
c.OSClient = osapi.NewOSServiceClient(c.conn)
c.MachineClient = machineapi.NewMachineServiceClient(c.conn)
c.TimeClient = timeapi.NewTimeServiceClient(c.conn)
c.NetworkClient = networkapi.NewNetworkServiceClient(c.conn)
@ -255,7 +252,6 @@ func NewClient(cfg *tls.Config, endpoints []string, port int, opts ...grpc.DialO
return
}
c.OSClient = osapi.NewOSServiceClient(c.conn)
c.MachineClient = machineapi.NewMachineServiceClient(c.conn)
c.TimeClient = timeapi.NewTimeServiceClient(c.conn)
c.NetworkClient = networkapi.NewNetworkServiceClient(c.conn)
@ -333,9 +329,9 @@ func (c *Client) Kubeconfig(ctx context.Context) ([]byte, error) {
}
// Stats implements the proto.OSClient interface.
func (c *Client) Stats(ctx context.Context, namespace string, driver common.ContainerDriver, callOptions ...grpc.CallOption) (resp *osapi.StatsResponse, err error) {
resp, err = c.OSClient.Stats(
ctx, &osapi.StatsRequest{
func (c *Client) Stats(ctx context.Context, namespace string, driver common.ContainerDriver, callOptions ...grpc.CallOption) (resp *machineapi.StatsResponse, err error) {
resp, err = c.MachineClient.Stats(
ctx, &machineapi.StatsRequest{
Namespace: namespace,
Driver: driver,
},
@ -344,16 +340,16 @@ func (c *Client) Stats(ctx context.Context, namespace string, driver common.Cont
var filtered interface{}
filtered, err = FilterMessages(resp, err)
resp, _ = filtered.(*osapi.StatsResponse) //nolint: errcheck
resp, _ = filtered.(*machineapi.StatsResponse) //nolint: errcheck
return
}
// Containers implements the proto.OSClient interface.
func (c *Client) Containers(ctx context.Context, namespace string, driver common.ContainerDriver, callOptions ...grpc.CallOption) (resp *osapi.ContainersResponse, err error) {
resp, err = c.OSClient.Containers(
func (c *Client) Containers(ctx context.Context, namespace string, driver common.ContainerDriver, callOptions ...grpc.CallOption) (resp *machineapi.ContainersResponse, err error) {
resp, err = c.MachineClient.Containers(
ctx,
&osapi.ContainersRequest{
&machineapi.ContainersRequest{
Namespace: namespace,
Driver: driver,
},
@ -362,14 +358,14 @@ func (c *Client) Containers(ctx context.Context, namespace string, driver common
var filtered interface{}
filtered, err = FilterMessages(resp, err)
resp, _ = filtered.(*osapi.ContainersResponse) //nolint: errcheck
resp, _ = filtered.(*machineapi.ContainersResponse) //nolint: errcheck
return
}
// Restart implements the proto.OSClient interface.
func (c *Client) Restart(ctx context.Context, namespace string, driver common.ContainerDriver, id string, callOptions ...grpc.CallOption) (err error) {
_, err = c.OSClient.Restart(ctx, &osapi.RestartRequest{
_, err = c.MachineClient.Restart(ctx, &machineapi.RestartRequest{
Id: id,
Namespace: namespace,
Driver: driver,
@ -415,8 +411,8 @@ func (c *Client) Shutdown(ctx context.Context) (err error) {
}
// Dmesg implements the proto.OSClient interface.
func (c *Client) Dmesg(ctx context.Context, follow, tail bool) (osapi.OSService_DmesgClient, error) {
return c.OSClient.Dmesg(ctx, &osapi.DmesgRequest{
func (c *Client) Dmesg(ctx context.Context, follow, tail bool) (machineapi.MachineService_DmesgClient, error) {
return c.MachineClient.Dmesg(ctx, &machineapi.DmesgRequest{
Follow: follow,
Tail: tail,
})
@ -481,8 +477,8 @@ func (c *Client) Interfaces(ctx context.Context, callOptions ...grpc.CallOption)
}
// Processes implements the proto.OSClient interface.
func (c *Client) Processes(ctx context.Context, callOptions ...grpc.CallOption) (resp *osapi.ProcessesResponse, err error) {
resp, err = c.OSClient.Processes(
func (c *Client) Processes(ctx context.Context, callOptions ...grpc.CallOption) (resp *machineapi.ProcessesResponse, err error) {
resp, err = c.MachineClient.Processes(
ctx,
&empty.Empty{},
callOptions...,
@ -490,14 +486,14 @@ func (c *Client) Processes(ctx context.Context, callOptions ...grpc.CallOption)
var filtered interface{}
filtered, err = FilterMessages(resp, err)
resp, _ = filtered.(*osapi.ProcessesResponse) //nolint: errcheck
resp, _ = filtered.(*machineapi.ProcessesResponse) //nolint: errcheck
return
}
// Memory implements the proto.OSClient interface.
func (c *Client) Memory(ctx context.Context, callOptions ...grpc.CallOption) (resp *osapi.MemoryResponse, err error) {
resp, err = c.OSClient.Memory(
func (c *Client) Memory(ctx context.Context, callOptions ...grpc.CallOption) (resp *machineapi.MemoryResponse, err error) {
resp, err = c.MachineClient.Memory(
ctx,
&empty.Empty{},
callOptions...,
@ -505,7 +501,7 @@ func (c *Client) Memory(ctx context.Context, callOptions ...grpc.CallOption) (re
var filtered interface{}
filtered, err = FilterMessages(resp, err)
resp, _ = filtered.(*osapi.MemoryResponse) //nolint: errcheck
resp, _ = filtered.(*machineapi.MemoryResponse) //nolint: errcheck
return
}

3
pkg/constants/constants.go
View File

@ -236,9 +236,6 @@ const (
// NetworkSocketPath is the path to file socket of network API.
NetworkSocketPath = SystemRunPath + "/networkd/networkd.sock"
// OSSocketPath is the path to file socket of os API.
OSSocketPath = SystemRunPath + "/osd/osd.sock"
// RouterdSocketPath is the path to file socket of router API.
RouterdSocketPath = SystemRunPath + "/routerd/routerd.sock"