add docker specifics
This commit is contained in:
74
local/service/docker_log_streamer.go
Normal file
74
local/service/docker_log_streamer.go
Normal file
@@ -0,0 +1,74 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"acc-server-manager/local/model"
|
||||
"acc-server-manager/local/utl/logging"
|
||||
"bufio"
|
||||
"context"
|
||||
"io"
|
||||
"sync"
|
||||
|
||||
"github.com/docker/docker/api/types"
|
||||
dockerclient "github.com/docker/docker/client"
|
||||
"github.com/docker/docker/pkg/stdcopy"
|
||||
"github.com/google/uuid"
|
||||
)
|
||||
|
||||
type DockerLogStreamer struct {
|
||||
client *dockerclient.Client
|
||||
cancels sync.Map // uuid.UUID → context.CancelFunc
|
||||
}
|
||||
|
||||
func NewDockerLogStreamer(client *dockerclient.Client) *DockerLogStreamer {
|
||||
return &DockerLogStreamer{client: client}
|
||||
}
|
||||
|
||||
func (s *DockerLogStreamer) Start(ctx context.Context, server *model.Server, handleLine func(string)) error {
|
||||
if _, exists := s.cancels.Load(server.ID); exists {
|
||||
return nil
|
||||
}
|
||||
if server.ContainerID == "" {
|
||||
logging.Warn("DockerLogStreamer.Start: no container ID for server %s, skipping", server.ID)
|
||||
return nil
|
||||
}
|
||||
|
||||
streamCtx, cancel := context.WithCancel(ctx)
|
||||
s.cancels.Store(server.ID, cancel)
|
||||
|
||||
go func() {
|
||||
defer s.cancels.Delete(server.ID)
|
||||
|
||||
rc, err := s.client.ContainerLogs(streamCtx, server.ContainerID, types.ContainerLogsOptions{
|
||||
ShowStdout: true,
|
||||
ShowStderr: true,
|
||||
Follow: true,
|
||||
Timestamps: false,
|
||||
})
|
||||
if err != nil {
|
||||
logging.Error("DockerLogStreamer: failed to attach to container %s: %v", server.ContainerID[:12], err)
|
||||
return
|
||||
}
|
||||
defer rc.Close()
|
||||
|
||||
pr, pw := io.Pipe()
|
||||
go func() {
|
||||
defer pw.Close()
|
||||
if _, err := stdcopy.StdCopy(pw, pw, rc); err != nil && err != io.EOF {
|
||||
logging.Warn("DockerLogStreamer: stdcopy error for %s: %v", server.ServiceName, err)
|
||||
}
|
||||
}()
|
||||
|
||||
scanner := bufio.NewScanner(pr)
|
||||
for scanner.Scan() {
|
||||
handleLine(scanner.Text())
|
||||
}
|
||||
}()
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *DockerLogStreamer) Stop(serverID uuid.UUID) {
|
||||
if v, ok := s.cancels.LoadAndDelete(serverID); ok {
|
||||
v.(context.CancelFunc)()
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user