mirror of
https://github.com/psviderski/uncloud.git
synced 2026-08-26 19:13:34 +00:00
feat(scale): store container spec in local machine.db sqlite
This commit is contained in:
@@ -0,0 +1,42 @@
|
||||
package machine
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
|
||||
"github.com/jmoiron/sqlx"
|
||||
_ "modernc.org/sqlite"
|
||||
)
|
||||
|
||||
const DBFileName = "machine.db"
|
||||
|
||||
// NewDB creates a new connection to machine SQLite database and runs schema migrations if necessary.
|
||||
func NewDB(path string) (*sqlx.DB, error) {
|
||||
// - Write-Ahead Logging (WAL) mode for better read/write performance.
|
||||
// - Busy timeout (5s) to make concurrent writes wait on each other instead of failing immediately.
|
||||
conn := path + "?_pragma=journal_mode=WAL&_pragma=synchronous=NORMAL&_pragma=busy_timeout=5000&_time_format=sqlite"
|
||||
db, err := sqlx.Connect("sqlite", conn)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("connect to SQLite database '%s': %w", conn, err)
|
||||
}
|
||||
|
||||
schema := `
|
||||
CREATE TABLE IF NOT EXISTS containers (
|
||||
id TEXT NOT NULL PRIMARY KEY,
|
||||
service_id TEXT NOT NULL,
|
||||
service_name TEXT AS (json_extract(service_spec, '$.Name')),
|
||||
service_spec TEXT NOT NULL CHECK (json_valid(service_spec)),
|
||||
-- 'subsecond' modifier is used to store timestamps with millisecond precision.
|
||||
created_at TIMESTAMP NOT NULL DEFAULT (datetime('subsecond')),
|
||||
updated_at TIMESTAMP NOT NULL DEFAULT (datetime('subsecond'))
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_containers_service_id ON containers (service_id);
|
||||
CREATE INDEX IF NOT EXISTS idx_containers_service_name ON containers (service_name);
|
||||
`
|
||||
|
||||
if _, err = db.Exec(schema); err != nil {
|
||||
return nil, fmt.Errorf("create schema: %w", err)
|
||||
}
|
||||
|
||||
return db, nil
|
||||
}
|
||||
@@ -5,6 +5,10 @@ import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"strconv"
|
||||
"strings"
|
||||
|
||||
"github.com/distribution/reference"
|
||||
"github.com/docker/docker/api/types"
|
||||
"github.com/docker/docker/api/types/container"
|
||||
@@ -16,6 +20,7 @@ import (
|
||||
"github.com/google/go-containerregistry/pkg/authn"
|
||||
"github.com/google/go-containerregistry/pkg/name"
|
||||
"github.com/google/go-containerregistry/pkg/v1/remote"
|
||||
"github.com/jmoiron/sqlx"
|
||||
"github.com/opencontainers/go-digest"
|
||||
ocispec "github.com/opencontainers/image-spec/specs-go/v1"
|
||||
"github.com/psviderski/uncloud/internal/machine/api/pb"
|
||||
@@ -25,20 +30,21 @@ import (
|
||||
"google.golang.org/grpc/codes"
|
||||
"google.golang.org/grpc/status"
|
||||
"google.golang.org/protobuf/types/known/emptypb"
|
||||
"io"
|
||||
"strconv"
|
||||
"strings"
|
||||
)
|
||||
|
||||
// Server implements the gRPC Docker service that proxies requests to the Docker daemon.
|
||||
type Server struct {
|
||||
pb.UnimplementedDockerServer
|
||||
client *client.Client
|
||||
db *sqlx.DB
|
||||
}
|
||||
|
||||
// NewServer creates a new Docker gRPC server with the provided Docker client.
|
||||
func NewServer(cli *client.Client) *Server {
|
||||
return &Server{client: cli}
|
||||
func NewServer(cli *client.Client, db *sqlx.DB) *Server {
|
||||
return &Server{
|
||||
client: cli,
|
||||
db: db,
|
||||
}
|
||||
}
|
||||
|
||||
// CreateContainer creates a new container based on the given configuration.
|
||||
@@ -424,5 +430,23 @@ func (s *Server) CreateServiceContainer(
|
||||
return nil, status.Errorf(codes.Internal, "marshal response: %v", err)
|
||||
}
|
||||
|
||||
// Store the container spec in the database or remove the container with its anonymous volumes if storing fails.
|
||||
removeContainer := func() {
|
||||
_ = s.client.ContainerRemove(ctx, resp.ID, container.RemoveOptions{RemoveVolumes: true})
|
||||
}
|
||||
|
||||
specBytes, err := json.Marshal(spec)
|
||||
if err != nil {
|
||||
removeContainer()
|
||||
return nil, status.Errorf(codes.Internal, "marshal service spec: %v", err)
|
||||
}
|
||||
|
||||
if _, err = s.db.ExecContext(ctx,
|
||||
`INSERT INTO containers (id, service_id, service_spec) VALUES ($1, $2, $3)`,
|
||||
resp.ID, req.ServiceId, string(specBytes)); err != nil {
|
||||
removeContainer()
|
||||
return nil, status.Errorf(codes.Internal, "store container in database: %v", err)
|
||||
}
|
||||
|
||||
return &pb.CreateContainerResponse{Response: respBytes}, nil
|
||||
}
|
||||
|
||||
@@ -5,14 +5,6 @@ import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"github.com/docker/docker/client"
|
||||
"github.com/docker/go-connections/sockets"
|
||||
"github.com/siderolabs/grpc-proxy/proxy"
|
||||
"golang.org/x/sync/errgroup"
|
||||
"google.golang.org/grpc"
|
||||
"google.golang.org/grpc/codes"
|
||||
"google.golang.org/grpc/status"
|
||||
"google.golang.org/protobuf/types/known/emptypb"
|
||||
"log/slog"
|
||||
"net"
|
||||
"net/netip"
|
||||
@@ -21,6 +13,9 @@ import (
|
||||
"path/filepath"
|
||||
"slices"
|
||||
"strconv"
|
||||
|
||||
"github.com/docker/docker/client"
|
||||
"github.com/docker/go-connections/sockets"
|
||||
"github.com/psviderski/uncloud/internal/corrosion"
|
||||
"github.com/psviderski/uncloud/internal/docker"
|
||||
"github.com/psviderski/uncloud/internal/fs"
|
||||
@@ -32,6 +27,12 @@ import (
|
||||
machinedocker "github.com/psviderski/uncloud/internal/machine/docker"
|
||||
"github.com/psviderski/uncloud/internal/machine/network"
|
||||
"github.com/psviderski/uncloud/internal/machine/store"
|
||||
"github.com/siderolabs/grpc-proxy/proxy"
|
||||
"golang.org/x/sync/errgroup"
|
||||
"google.golang.org/grpc"
|
||||
"google.golang.org/grpc/codes"
|
||||
"google.golang.org/grpc/status"
|
||||
"google.golang.org/protobuf/types/known/emptypb"
|
||||
)
|
||||
|
||||
const (
|
||||
@@ -209,7 +210,13 @@ func NewMachine(config *Config) (*Machine, error) {
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("create Docker client: %w", err)
|
||||
}
|
||||
dockerServer := machinedocker.NewServer(dockerCli)
|
||||
|
||||
dbFilePath := filepath.Join(config.DataDir, DBFileName)
|
||||
db, err := NewDB(dbFilePath)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("init machine database: %w", err)
|
||||
}
|
||||
dockerServer := machinedocker.NewServer(dockerCli, db)
|
||||
|
||||
// Init a local gRPC proxy server that proxies requests to the local or remote machine API servers.
|
||||
proxyDirector := apiproxy.NewDirector(config.MachineSockPath, APIPort)
|
||||
|
||||
Reference in New Issue
Block a user