From 72346151105d4d59ab89adf728ed00d08dc98efe Mon Sep 17 00:00:00 2001 From: n/a Date: Wed, 12 Aug 2026 22:48:00 +0200 Subject: [PATCH 1/2] command listener removed --- go.mod | 2 +- go.sum | 2 + internal/apiclient/client.go | 12 +++ internal/grpc/command.go | 190 ----------------------------------- internal/grpc/grpcapi.go | 3 - internal/server/restapi.go | 13 ++- 6 files changed, 27 insertions(+), 195 deletions(-) delete mode 100644 internal/grpc/command.go diff --git a/go.mod b/go.mod index 4126099..485f594 100644 --- a/go.mod +++ b/go.mod @@ -8,7 +8,7 @@ require ( github.com/fatih/color v1.18.0 github.com/google/shlex v0.0.0-20191202100458-e7afc7fbc510 github.com/gorilla/mux v1.8.1 - github.com/k8shell-io/common v0.37.0 + github.com/k8shell-io/common v0.39.0 github.com/k8shell-io/k8shell-go v0.2.3 github.com/pkg/sftp v1.13.10 github.com/rs/zerolog v1.34.0 diff --git a/go.sum b/go.sum index f4829e4..3fce75b 100644 --- a/go.sum +++ b/go.sum @@ -46,6 +46,8 @@ github.com/k8shell-io/common v0.36.0 h1:fkMH1XfYRLzDxqhIq5/luHusWWPGnCGJUXSTEIhE github.com/k8shell-io/common v0.36.0/go.mod h1:E8dsb9ta4v3ne61AJgtRyTTbTkMMmKeCMAcXD+/9+cY= github.com/k8shell-io/common v0.37.0 h1:whq66WosIJECKErUKZF1RQep7tdpOfI6GtP4hXREpsQ= github.com/k8shell-io/common v0.37.0/go.mod h1:E8dsb9ta4v3ne61AJgtRyTTbTkMMmKeCMAcXD+/9+cY= +github.com/k8shell-io/common v0.39.0 h1:hfrKZYX2lBonornGrfK35rSY/+5o2sK+SakgUPKDL24= +github.com/k8shell-io/common v0.39.0/go.mod h1:E8dsb9ta4v3ne61AJgtRyTTbTkMMmKeCMAcXD+/9+cY= github.com/k8shell-io/k8shell-go v0.2.1 h1:6n88ijXkzP39//lIy4ai3XqtpSUXzoa/dVaWogHQYf4= github.com/k8shell-io/k8shell-go v0.2.1/go.mod h1:j1JHgUIKIbaiRaitx6Pzw37ahqS4Hu9OcM4uvJ7BP4g= github.com/k8shell-io/k8shell-go v0.2.2 h1:rwLOeIfyq1+l2Jyv0ak/lXZS7x6xbA5ye0yMsRkTonw= diff --git a/internal/apiclient/client.go b/internal/apiclient/client.go index 06a0984..6eccc2a 100644 --- a/internal/apiclient/client.go +++ b/internal/apiclient/client.go @@ -69,3 +69,15 @@ func (c *Client) GetUserProfile(ctx context.Context, username string) (*models.U func (c *Client) SetUserPassword(ctx context.Context, username, password, currentPassword string) (*models.User, error) { return c.sdk.SetUserPassword(ctx, username, password, currentPassword) } + +// StopWorkspace shuts down the named workspace, keeping its persistent +// storage intact so it can later be resumed with StartWorkspace. +func (c *Client) StopWorkspace(ctx context.Context, name string) error { + return c.sdk.DeleteWorkspace(ctx, name, false) +} + +// TerminateWorkspace shuts down the named workspace and permanently deletes +// its persistent storage. +func (c *Client) TerminateWorkspace(ctx context.Context, name string) error { + return c.sdk.DeleteWorkspace(ctx, name, true) +} diff --git a/internal/grpc/command.go b/internal/grpc/command.go deleted file mode 100644 index d783e24..0000000 --- a/internal/grpc/command.go +++ /dev/null @@ -1,190 +0,0 @@ -// Use of this source code is governed by a AGPLv3 -// license that can be found in the LICENSE file. - -package grpc - -import ( - "context" - "errors" - "io" - "strconv" - "strings" - "sync" - "time" - - k8shelldv1 "github.com/k8shell-io/common/pkg/api/gen/go/k8shelld/v1" - "github.com/k8shell-io/k8shelld/internal/logger" - "github.com/rs/zerolog" - "google.golang.org/grpc/codes" - "google.golang.org/grpc/status" -) - -// CommandServiceServer implements the gRPC server for command execution. -// -// It manages a single active logical connection (bidi stream) from a client. -// Other parts of the server can send commands to that client via SendCommand, -// and await the reply. Replies from the client are correlated using command_id. -type CommandServiceServer struct { - k8shelldv1.UnimplementedCommandServiceServer - - logger *zerolog.Logger - - mu sync.Mutex - clients map[uint64]chan *k8shelldv1.CommandMessage - nextClient uint64 - pending map[string]chan string -} - -// NewCommandServiceServer creates a new CommandServiceServer instance. -func NewCommandServiceServer() *CommandServiceServer { - return &CommandServiceServer{ - logger: logger.NewLogger("grpc-commands"), - clients: make(map[uint64]chan *k8shelldv1.CommandMessage), - pending: make(map[string]chan string), - } -} - -// CommandListener implements the bidi streaming RPC defined in the proto. -// Only a single active client is supported at a time; a new connection replaces any previous one. -func (s *CommandServiceServer) CommandListener(stream k8shelldv1.CommandService_CommandListenerServer) error { - ctx := stream.Context() - - // Create a send channel dedicated to this stream instance and register client. - ch := make(chan *k8shelldv1.CommandMessage, 16) - - s.mu.Lock() - clientID := s.nextClient - s.nextClient++ - s.clients[clientID] = ch - s.mu.Unlock() - - s.logger.Info().Uint64("client_id", clientID).Msg("command listener connected") - - done := make(chan struct{}) - go func() { - defer close(done) - for msg := range ch { - if err := stream.Send(msg); err != nil { - if errors.Is(err, context.Canceled) || status.Code(err) == codes.Canceled { - s.logger.Debug().Uint64("client_id", clientID).Msg("command stream send canceled by client") - } else { - s.logger.Error().Uint64("client_id", clientID).Err(err).Msg("failed to send command to client") - } - return - } - } - }() - - for { - in, err := stream.Recv() - if err != nil { - if err == io.EOF { - break - } - if errors.Is(err, context.Canceled) || status.Code(err) == codes.Canceled { - s.logger.Debug().Uint64("client_id", clientID).Msg("command stream canceled by client") - break - } - s.logger.Error().Uint64("client_id", clientID).Err(err).Msg("error receiving command reply from client") - break - } - - cmdID := in.GetCommandId() - switch payload := in.Payload.(type) { - case *k8shelldv1.CommandMessage_Reply: - reply := payload.Reply - - s.mu.Lock() - chReply, ok := s.pending[cmdID] - if ok { - delete(s.pending, cmdID) - } - s.mu.Unlock() - - if ok { - select { - case chReply <- reply: - close(chReply) - default: - // receiver gone; drop reply - } - } else { - s.logger.Warn().Str("command_id", cmdID).Uint64("client_id", clientID). - Msg("received reply for unknown command_id") - } - default: - s.logger.Warn().Uint64("client_id", clientID). - Msg("received unexpected command payload type from client; ignoring") - } - } - - s.mu.Lock() - if chCurrent, ok := s.clients[clientID]; ok && chCurrent == ch { - close(ch) - delete(s.clients, clientID) - } - s.mu.Unlock() - - select { - case <-done: - case <-ctx.Done(): - } - - s.logger.Info().Uint64("client_id", clientID).Msg("command listener disconnected") - return nil -} - -// SendCommand sends a command to exactly one connected client and waits for reply. -func (s *CommandServiceServer) SendCommand(ctx context.Context, command string) (string, error) { - if command == "" { - return "", errors.New("empty command") - } - - s.mu.Lock() - // pick any one client (first in map) to send the command to - var sendCh chan *k8shelldv1.CommandMessage - for _, ch := range s.clients { - sendCh = ch - break - } - if sendCh == nil { - s.mu.Unlock() - return "", status.Errorf(codes.Unavailable, "no command client connected") - } - - cmdID := strconv.FormatInt(time.Now().UnixNano(), 10) - replyCh := make(chan string, 1) - s.pending[cmdID] = replyCh - s.mu.Unlock() - - msg := &k8shelldv1.CommandMessage{ - CommandId: cmdID, - Payload: &k8shelldv1.CommandMessage_Command{ - Command: command, - }, - } - - // Send the command to the selected client. - select { - case sendCh <- msg: - case <-ctx.Done(): - s.mu.Lock() - delete(s.pending, cmdID) - s.mu.Unlock() - return "", ctx.Err() - } - - // Wait for the reply or cancellation. - select { - case reply := <-replyCh: - if strings.HasPrefix(reply, "error:") { - return "", errors.New(strings.TrimSpace(strings.TrimPrefix(reply, "error:"))) - } - return reply, nil - case <-ctx.Done(): - s.mu.Lock() - delete(s.pending, cmdID) - s.mu.Unlock() - return "", ctx.Err() - } -} diff --git a/internal/grpc/grpcapi.go b/internal/grpc/grpcapi.go index fb78a00..5226bb1 100644 --- a/internal/grpc/grpcapi.go +++ b/internal/grpc/grpcapi.go @@ -56,7 +56,6 @@ type GRPCService struct { UnixSocketStore *sync.Map // The store for the unix socket data apiClientx *apiclient.Client // The API client to communicate with the API server appManager *apps.AppManager // The app manager - CommandService *CommandServiceServer // The command service sysInfo *system.SystemInfo // The system information detachedSessionTTL time.Duration // max TTL for sessions with no client; 0 = no GC allowSessionDetach bool // whether clients may detach/attach PTY sessions @@ -127,7 +126,6 @@ func NewGRPCService(config *config.Config, blueprint *commonmodels.Blueprint, us UnixSocketStore: &sync.Map{}, apiClientx: apiClient, appManager: appManager, - CommandService: NewCommandServiceServer(), sysInfo: sysInfo, detachedSessionTTL: detachedTTL, allowSessionDetach: config.Shells.AllowSessionDetach, @@ -170,7 +168,6 @@ func (a *GRPCService) Serve(ctx context.Context) error { k8shelldv1.RegisterSystemServiceServer(s, NewSystemServiceServer(a)) k8shelldv1.RegisterSshServiceServer(s, NewSshServiceServer(a)) k8shelldv1.RegisterAppServiceServer(s, NewAppServiceServer(a.appManager)) - k8shelldv1.RegisterCommandServiceServer(s, a.CommandService) a.logger.Info().Msgf("GRPC services server registered") return nil }); err != nil { diff --git a/internal/server/restapi.go b/internal/server/restapi.go index b3240c6..eed1185 100644 --- a/internal/server/restapi.go +++ b/internal/server/restapi.go @@ -188,6 +188,11 @@ func (a *RESTService) GetSessions(w http.ResponseWriter, r *http.Request) { } func (a *RESTService) Shutdown(w http.ResponseWriter, r *http.Request) { + if a.server.apiClientx == nil { + http.Error(w, "API server not configured.", http.StatusServiceUnavailable) + return + } + action := r.URL.Query().Get("action") if action == "" { action = "stop" @@ -197,7 +202,13 @@ func (a *RESTService) Shutdown(w http.ResponseWriter, r *http.Request) { return } a.logger.Debug().Msgf("Shutting down workspace %s (action=%s)", a.server.workspace, action) - _, err := a.server.grpcService.CommandService.SendCommand(r.Context(), "shutdown "+action) + + var err error + if action == "delete" { + err = a.server.apiClientx.TerminateWorkspace(r.Context(), a.server.workspace) + } else { + err = a.server.apiClientx.StopWorkspace(r.Context(), a.server.workspace) + } if err != nil { a.logger.Warn().Msgf("Cannot shutdown workspace: %v", err) http.Error(w, "Failed to shutdown workspace", http.StatusBadGateway) From 4ffbb1032c21312bda4b75e6d41328e58806fb69 Mon Sep 17 00:00:00 2001 From: n/a Date: Mon, 17 Aug 2026 23:27:18 +0200 Subject: [PATCH 2/2] common v bump --- go.mod | 2 +- go.sum | 2 ++ 2 files changed, 3 insertions(+), 1 deletion(-) diff --git a/go.mod b/go.mod index 485f594..2594d92 100644 --- a/go.mod +++ b/go.mod @@ -8,7 +8,7 @@ require ( github.com/fatih/color v1.18.0 github.com/google/shlex v0.0.0-20191202100458-e7afc7fbc510 github.com/gorilla/mux v1.8.1 - github.com/k8shell-io/common v0.39.0 + github.com/k8shell-io/common v0.40.0 github.com/k8shell-io/k8shell-go v0.2.3 github.com/pkg/sftp v1.13.10 github.com/rs/zerolog v1.34.0 diff --git a/go.sum b/go.sum index 3fce75b..6c8664a 100644 --- a/go.sum +++ b/go.sum @@ -48,6 +48,8 @@ github.com/k8shell-io/common v0.37.0 h1:whq66WosIJECKErUKZF1RQep7tdpOfI6GtP4hXRE github.com/k8shell-io/common v0.37.0/go.mod h1:E8dsb9ta4v3ne61AJgtRyTTbTkMMmKeCMAcXD+/9+cY= github.com/k8shell-io/common v0.39.0 h1:hfrKZYX2lBonornGrfK35rSY/+5o2sK+SakgUPKDL24= github.com/k8shell-io/common v0.39.0/go.mod h1:E8dsb9ta4v3ne61AJgtRyTTbTkMMmKeCMAcXD+/9+cY= +github.com/k8shell-io/common v0.40.0 h1:MhQPVI5oe+JSdRJhWrtvCTJf0MaJDjM58KR7Nq3+lsM= +github.com/k8shell-io/common v0.40.0/go.mod h1:E8dsb9ta4v3ne61AJgtRyTTbTkMMmKeCMAcXD+/9+cY= github.com/k8shell-io/k8shell-go v0.2.1 h1:6n88ijXkzP39//lIy4ai3XqtpSUXzoa/dVaWogHQYf4= github.com/k8shell-io/k8shell-go v0.2.1/go.mod h1:j1JHgUIKIbaiRaitx6Pzw37ahqS4Hu9OcM4uvJ7BP4g= github.com/k8shell-io/k8shell-go v0.2.2 h1:rwLOeIfyq1+l2Jyv0ak/lXZS7x6xbA5ye0yMsRkTonw=