403 lines
12 KiB
Go
403 lines
12 KiB
Go
package api
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"net/http"
|
|
"strings"
|
|
"time"
|
|
|
|
"browser.local/platform/domain"
|
|
"browser.local/platform/dto"
|
|
"browser.local/platform/service"
|
|
)
|
|
|
|
type componentLogServerContextKey struct{}
|
|
|
|
const (
|
|
logEventHeartbeatInterval = 15 * time.Second
|
|
)
|
|
|
|
// serverLogEvents streams platform-accepted live append events for the terminal drawer.
|
|
func (h *coreHandlers) serverLogEvents(w http.ResponseWriter, r *http.Request) {
|
|
if r.Method != http.MethodGet {
|
|
writeMethodNotAllowed(w, http.MethodGet)
|
|
return
|
|
}
|
|
jobID := strings.TrimSpace(r.URL.Query().Get("jobId"))
|
|
instance, streams, subscription, err := h.openLogEventSubscription(r, jobID)
|
|
if err != nil {
|
|
writeServiceError(w, err)
|
|
return
|
|
}
|
|
defer subscription.Close()
|
|
flusher, ok := w.(http.Flusher)
|
|
if !ok {
|
|
writeServiceError(w, fmt.Errorf("streaming response unsupported"))
|
|
return
|
|
}
|
|
|
|
header := w.Header()
|
|
header.Set("Content-Type", "text/event-stream")
|
|
header.Set("Cache-Control", "no-cache, no-transform")
|
|
header.Set("Connection", "keep-alive")
|
|
header.Set("X-Accel-Buffering", "no")
|
|
w.WriteHeader(http.StatusOK)
|
|
|
|
active := supervisedLogSession{}
|
|
if jobID != "" {
|
|
active = activeJobLogSession(jobID, streams)
|
|
} else if isComponentLogRequest(r) {
|
|
active = activeComponentLogSession(streams)
|
|
} else {
|
|
active = activeSupervisedLogSession(streams)
|
|
}
|
|
emittedThrough, err := h.writeCurrentLogSession(w, instance.ID, active)
|
|
if err != nil {
|
|
_ = writeSSEJSON(w, "error", "", map[string]string{"message": err.Error()})
|
|
flusher.Flush()
|
|
return
|
|
}
|
|
if err := writeSSEJSON(w, "ready", "", dto.LogStreamEventsReadyResponse{ServerInstanceID: instance.ID, StreamCount: len(active.streams), ServerTime: time.Now().UTC()}); err != nil {
|
|
return
|
|
}
|
|
flusher.Flush()
|
|
|
|
heartbeat := time.NewTicker(logEventHeartbeatInterval)
|
|
defer heartbeat.Stop()
|
|
for {
|
|
select {
|
|
case <-r.Context().Done():
|
|
return
|
|
case subscriptionEvent, ok := <-subscription.Events:
|
|
if !ok {
|
|
return
|
|
}
|
|
if subscriptionEvent.Kind == service.LogEventSubscriptionEventProcessState {
|
|
if jobID != "" {
|
|
continue
|
|
}
|
|
if isComponentLogRequest(r) {
|
|
continue
|
|
}
|
|
if subscriptionEvent.ServerInstanceID != instance.ID {
|
|
continue
|
|
}
|
|
if subscriptionEvent.ProcessState != domain.ServerInstanceStateRunning {
|
|
if active.sessionID == "" {
|
|
continue
|
|
}
|
|
active = supervisedLogSession{}
|
|
emittedThrough, err = h.writeCurrentLogSession(w, instance.ID, active)
|
|
if err != nil {
|
|
return
|
|
}
|
|
flusher.Flush()
|
|
continue
|
|
}
|
|
streams, err = h.loadLiveLogSnapshot(instance.ID, isComponentLogRequest(r), jobID)
|
|
if err != nil {
|
|
return
|
|
}
|
|
next := activeSupervisedLogSession(streams)
|
|
if sameSupervisedLogSession(active, next) {
|
|
continue
|
|
}
|
|
active = next
|
|
emittedThrough, err = h.writeCurrentLogSession(w, instance.ID, active)
|
|
if err != nil {
|
|
return
|
|
}
|
|
flusher.Flush()
|
|
continue
|
|
}
|
|
if subscriptionEvent.Kind != service.LogEventSubscriptionEventLog {
|
|
continue
|
|
}
|
|
event := subscriptionEvent.LogEvent
|
|
candidate := activeSupervisedLogSession([]domain.LogStream{event.Stream})
|
|
if jobID != "" {
|
|
if !strings.HasPrefix(event.Stream.ID, "job."+jobID+".") {
|
|
continue
|
|
}
|
|
candidate = activeJobLogSession(jobID, []domain.LogStream{event.Stream})
|
|
}
|
|
if active.allStreams {
|
|
candidate = supervisedLogSession{}
|
|
}
|
|
if jobID == "" && candidate.sessionID != "" && newerLogSession(candidate, active) {
|
|
next := candidate
|
|
if !sameSupervisedLogSession(active, next) {
|
|
active = next
|
|
emittedThrough, err = h.writeCurrentLogSession(w, instance.ID, active)
|
|
if err != nil {
|
|
return
|
|
}
|
|
if event.Entry.Seq > 0 && emittedThrough[event.Stream.ID] >= event.Entry.Seq {
|
|
emittedThrough[event.Stream.ID] = event.Entry.Seq - 1
|
|
}
|
|
flusher.Flush()
|
|
}
|
|
}
|
|
if !active.contains(event.Stream) {
|
|
if jobID == "" {
|
|
if !active.allStreams || event.Stream.ServerInstanceID != instance.ID {
|
|
continue
|
|
}
|
|
} else if !strings.HasPrefix(event.Stream.ID, "job."+jobID+".") {
|
|
continue
|
|
}
|
|
active.streams = append(active.streams, event.Stream)
|
|
if err := writeSSEJSON(w, "stream", "", dto.LogStreamFromDomain(event.Stream)); err != nil {
|
|
return
|
|
}
|
|
}
|
|
if !active.hasStream(event.Stream.ID) {
|
|
active.streams = append(active.streams, event.Stream)
|
|
if err := writeSSEJSON(w, "stream", "", dto.LogStreamFromDomain(event.Stream)); err != nil {
|
|
return
|
|
}
|
|
}
|
|
if event.Entry.Seq <= emittedThrough[event.Stream.ID] {
|
|
continue
|
|
}
|
|
if err := writeSSEJSON(w, "log", logEventID(event), dto.LogStreamEventFromDomain(event)); err != nil {
|
|
return
|
|
}
|
|
emittedThrough[event.Stream.ID] = event.Entry.Seq
|
|
flusher.Flush()
|
|
case <-heartbeat.C:
|
|
if _, err := fmt.Fprintf(w, ": heartbeat %s\n\n", time.Now().UTC().Format(time.RFC3339)); err != nil {
|
|
return
|
|
}
|
|
flusher.Flush()
|
|
}
|
|
}
|
|
}
|
|
|
|
type supervisedLogSession struct {
|
|
jobID string
|
|
sessionID string
|
|
startedAt time.Time
|
|
streams []domain.LogStream
|
|
allStreams bool
|
|
}
|
|
|
|
func activeComponentLogSession(streams []domain.LogStream) supervisedLogSession {
|
|
return supervisedLogSession{sessionID: "component", streams: append([]domain.LogStream(nil), streams...), allStreams: true}
|
|
}
|
|
|
|
func activeSupervisedLogSession(streams []domain.LogStream) supervisedLogSession {
|
|
active := supervisedLogSession{}
|
|
for _, stream := range streams {
|
|
if stream.Source != domain.LogStreamSourceProcess || strings.TrimSpace(stream.LogSessionID) == "" || stream.SessionStartedAt.IsZero() {
|
|
continue
|
|
}
|
|
candidate := supervisedLogSession{sessionID: stream.LogSessionID, startedAt: stream.SessionStartedAt}
|
|
if active.sessionID == "" || newerLogSession(candidate, active) {
|
|
active = candidate
|
|
}
|
|
}
|
|
if active.sessionID == "" {
|
|
return active
|
|
}
|
|
for _, stream := range streams {
|
|
if stream.Source == domain.LogStreamSourceProcess && stream.LogSessionID == active.sessionID && stream.SessionStartedAt.Equal(active.startedAt) {
|
|
active.streams = append(active.streams, stream)
|
|
}
|
|
}
|
|
return active
|
|
}
|
|
|
|
func activeJobLogSession(jobID string, streams []domain.LogStream) supervisedLogSession {
|
|
active := supervisedLogSession{jobID: jobID, sessionID: "job:" + jobID, streams: append([]domain.LogStream(nil), streams...)}
|
|
return active
|
|
}
|
|
|
|
func newerLogSession(candidate supervisedLogSession, current supervisedLogSession) bool {
|
|
if candidate.sessionID == "" || candidate.sessionID == current.sessionID {
|
|
return false
|
|
}
|
|
if current.sessionID == "" {
|
|
return true
|
|
}
|
|
if !candidate.startedAt.Equal(current.startedAt) {
|
|
return candidate.startedAt.After(current.startedAt)
|
|
}
|
|
return candidate.sessionID > current.sessionID
|
|
}
|
|
|
|
func sameSupervisedLogSession(left supervisedLogSession, right supervisedLogSession) bool {
|
|
return left.jobID == right.jobID && left.sessionID == right.sessionID && left.startedAt.Equal(right.startedAt)
|
|
}
|
|
|
|
func (session supervisedLogSession) contains(stream domain.LogStream) bool {
|
|
if session.jobID != "" {
|
|
return strings.HasPrefix(stream.ID, "job."+session.jobID+".") && session.hasStream(stream.ID)
|
|
}
|
|
if session.allStreams {
|
|
return session.hasStream(stream.ID)
|
|
}
|
|
return session.sessionID != "" && stream.Source == domain.LogStreamSourceProcess && stream.LogSessionID == session.sessionID && stream.SessionStartedAt.Equal(session.startedAt)
|
|
}
|
|
|
|
func (session supervisedLogSession) hasStream(streamID string) bool {
|
|
for _, stream := range session.streams {
|
|
if stream.ID == streamID {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
func (h *coreHandlers) writeCurrentLogSession(w http.ResponseWriter, serverInstanceID string, active supervisedLogSession) (map[string]uint64, error) {
|
|
emittedThrough := make(map[string]uint64, len(active.streams))
|
|
if err := writeSSEJSON(w, "session", "", dto.LogStreamEventsSessionResponse{ServerInstanceID: serverInstanceID, LogSessionID: active.sessionID, SessionStartedAt: active.startedAt, StreamCount: len(active.streams), ServerTime: time.Now().UTC()}); err != nil {
|
|
return nil, err
|
|
}
|
|
for _, stream := range active.streams {
|
|
emittedThrough[stream.ID] = stream.LatestSeq
|
|
if err := writeSSEJSON(w, "stream", "", dto.LogStreamFromDomain(stream)); err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
return emittedThrough, nil
|
|
}
|
|
|
|
func (h *coreHandlers) openLogEventSubscription(r *http.Request, jobID string) (domain.ServerInstance, []domain.LogStream, service.LogEventSubscription, error) {
|
|
var instance domain.ServerInstance
|
|
var streams []domain.LogStream
|
|
var subscription service.LogEventSubscription
|
|
var err error
|
|
if serverInstanceID, ok := r.Context().Value(componentLogServerContextKey{}).(string); ok && strings.TrimSpace(serverInstanceID) != "" {
|
|
instance, err = h.core.GetServerInstance(serverInstanceID)
|
|
if err == nil {
|
|
subscription, err = h.core.SubscribeLogEvents(instance.ID)
|
|
}
|
|
} else if h.enforceAuthorization {
|
|
sessionID := bearerToken(r)
|
|
instance, err = h.core.GetServerInstanceForSession(sessionID, r.PathValue("id"))
|
|
if err != nil {
|
|
return domain.ServerInstance{}, nil, subscription, err
|
|
}
|
|
subscription, err = h.core.SubscribeLogEventsForSession(sessionID, instance.ID)
|
|
if err != nil {
|
|
return domain.ServerInstance{}, nil, subscription, err
|
|
}
|
|
} else {
|
|
instance, err = h.core.GetServerInstance(r.PathValue("id"))
|
|
if err != nil {
|
|
return domain.ServerInstance{}, nil, subscription, err
|
|
}
|
|
subscription, err = h.core.SubscribeLogEvents(instance.ID)
|
|
if err != nil {
|
|
return domain.ServerInstance{}, nil, subscription, err
|
|
}
|
|
}
|
|
if err == nil && jobID != "" {
|
|
if h.enforceAuthorization && isComponentLogRequest(r) {
|
|
err = fmt.Errorf("job log streams require session authorization")
|
|
} else {
|
|
var job domain.Job
|
|
if h.enforceAuthorization {
|
|
job, err = h.core.GetJobForSession(bearerToken(r), jobID)
|
|
} else {
|
|
job, err = h.core.GetJob(jobID)
|
|
}
|
|
if err == nil && job.ServerInstanceID != instance.ID {
|
|
err = service.ErrForbidden
|
|
}
|
|
}
|
|
}
|
|
if err == nil {
|
|
streams, err = h.loadLiveLogSnapshot(instance.ID, isComponentLogRequest(r), jobID)
|
|
}
|
|
if err != nil && subscription.Close != nil {
|
|
subscription.Close()
|
|
}
|
|
return instance, streams, subscription, err
|
|
}
|
|
|
|
func withComponentLogServer(r *http.Request, serverInstanceID string) *http.Request {
|
|
return r.WithContext(context.WithValue(r.Context(), componentLogServerContextKey{}, serverInstanceID))
|
|
}
|
|
|
|
func isComponentLogRequest(r *http.Request) bool {
|
|
_, ok := r.Context().Value(componentLogServerContextKey{}).(string)
|
|
return ok
|
|
}
|
|
|
|
func (h *coreHandlers) loadLiveLogSnapshot(serverInstanceID string, includeDeclaredStreams bool, jobID string) ([]domain.LogStream, error) {
|
|
instance, err := h.core.GetServerInstance(serverInstanceID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if jobID == "" {
|
|
if strings.TrimSpace(instance.RunEndpointID) == "" {
|
|
return nil, nil
|
|
}
|
|
if !includeDeclaredStreams && instance.State != domain.ServerInstanceStateRunning {
|
|
return nil, nil
|
|
}
|
|
endpoint, err := h.core.GetRunEndpoint(instance.RunEndpointID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if endpoint.Status != domain.RunEndpointStatusOnline {
|
|
return nil, nil
|
|
}
|
|
}
|
|
streams, err := h.core.ListLogStreams(domain.LogStreamFilter{ServerInstanceID: instance.ID})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
current := make([]domain.LogStream, 0, len(streams))
|
|
for _, stream := range streams {
|
|
if jobID != "" {
|
|
if strings.HasPrefix(stream.ID, "job."+jobID+".") {
|
|
current = append(current, stream)
|
|
}
|
|
continue
|
|
}
|
|
if includeDeclaredStreams {
|
|
if stream.Source != domain.LogStreamSourceProcess && stream.Source != domain.LogStreamSourceFile && stream.Source != domain.LogStreamSourceManagementProgram {
|
|
continue
|
|
}
|
|
current = append(current, stream)
|
|
continue
|
|
}
|
|
if stream.Source == domain.LogStreamSourceProcess && strings.TrimSpace(stream.LogSessionID) != "" && !stream.SessionStartedAt.IsZero() {
|
|
current = append(current, stream)
|
|
}
|
|
}
|
|
return current, nil
|
|
}
|
|
|
|
func writeSSEJSON(w http.ResponseWriter, eventName string, id string, value any) error {
|
|
payload, err := json.Marshal(value)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if id != "" {
|
|
if _, err := fmt.Fprintf(w, "id: %s\n", sanitizeSSEField(id)); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
if _, err := fmt.Fprintf(w, "event: %s\n", sanitizeSSEField(eventName)); err != nil {
|
|
return err
|
|
}
|
|
_, err = fmt.Fprintf(w, "data: %s\n\n", payload)
|
|
return err
|
|
}
|
|
|
|
func sanitizeSSEField(value string) string {
|
|
value = strings.ReplaceAll(value, "\r", "")
|
|
value = strings.ReplaceAll(value, "\n", "")
|
|
return value
|
|
}
|
|
|
|
func logEventID(event domain.LogStreamEvent) string {
|
|
return fmt.Sprintf("%s:%d", sanitizeSSEField(event.Stream.ID), event.Entry.Seq)
|
|
}
|