Reduce log ingest and SCUM page refresh pressure
This commit is contained in:
@@ -41,6 +41,7 @@ type StoreSnapshot struct {
|
||||
type runtimeSnapshot struct {
|
||||
RunControlSessions []domain.RunControlSession `json:"runControlSessions"`
|
||||
RunEndpoints []domain.RunEndpoint `json:"runEndpoints"`
|
||||
LogStreams []domain.LogStream `json:"logStreams"`
|
||||
}
|
||||
|
||||
type FileStore struct {
|
||||
@@ -131,7 +132,7 @@ func (store *FileStore) RunUpdateJobs() RunUpdateJobRepository {
|
||||
}
|
||||
|
||||
func (store *FileStore) LogStreams() LogStreamRepository {
|
||||
return &persistentRepository[domain.LogStream, domain.LogStreamFilter]{repository: store.MemoryStore.logStreams, persist: store.persist}
|
||||
return &persistentRepository[domain.LogStream, domain.LogStreamFilter]{repository: store.MemoryStore.logStreams, persist: store.persistRuntime}
|
||||
}
|
||||
|
||||
func (store *FileStore) MetricSamples() MetricSampleRepository {
|
||||
@@ -318,6 +319,7 @@ func (store *FileStore) runtimeSnapshot() runtimeSnapshot {
|
||||
return runtimeSnapshot{
|
||||
RunControlSessions: snapshotRepository(store.MemoryStore.runSessions),
|
||||
RunEndpoints: snapshotRepository(store.MemoryStore.runEndpoints),
|
||||
LogStreams: snapshotRepository(store.MemoryStore.logStreams),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -350,6 +352,7 @@ func (store *FileStore) loadSnapshot(snapshot StoreSnapshot) {
|
||||
func (store *FileStore) loadRuntimeSnapshot(snapshot runtimeSnapshot) {
|
||||
loadRepository(store.MemoryStore.runSessions, snapshot.RunControlSessions)
|
||||
loadRepository(store.MemoryStore.runEndpoints, snapshot.RunEndpoints)
|
||||
loadRepository(store.MemoryStore.logStreams, snapshot.LogStreams)
|
||||
}
|
||||
|
||||
type mutableRepository[T any, F any] interface {
|
||||
|
||||
@@ -114,7 +114,7 @@ func (store *MySQLStore) RunUpdateJobs() RunUpdateJobRepository {
|
||||
}
|
||||
|
||||
func (store *MySQLStore) LogStreams() LogStreamRepository {
|
||||
return &persistentRepository[domain.LogStream, domain.LogStreamFilter]{repository: store.MemoryStore.logStreams, persist: store.persist}
|
||||
return &persistentRepository[domain.LogStream, domain.LogStreamFilter]{repository: store.MemoryStore.logStreams, persist: store.persistRuntime}
|
||||
}
|
||||
|
||||
func (store *MySQLStore) MetricSamples() MetricSampleRepository {
|
||||
@@ -280,6 +280,7 @@ func (store *MySQLStore) runtimeSnapshot() runtimeSnapshot {
|
||||
return runtimeSnapshot{
|
||||
RunControlSessions: snapshotRepository(store.MemoryStore.runSessions),
|
||||
RunEndpoints: snapshotRepository(store.MemoryStore.runEndpoints),
|
||||
LogStreams: snapshotRepository(store.MemoryStore.logStreams),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -312,4 +313,5 @@ func (store *MySQLStore) loadSnapshot(snapshot StoreSnapshot) {
|
||||
func (store *MySQLStore) loadRuntimeSnapshot(snapshot runtimeSnapshot) {
|
||||
loadRepository(store.MemoryStore.runSessions, snapshot.RunControlSessions)
|
||||
loadRepository(store.MemoryStore.runEndpoints, snapshot.RunEndpoints)
|
||||
loadRepository(store.MemoryStore.logStreams, snapshot.LogStreams)
|
||||
}
|
||||
|
||||
@@ -330,6 +330,10 @@ func TestFileStorePersistsRunRuntimeStateSeparately(t *testing.T) {
|
||||
if err := store.RunControlSessions().Create(session); err != nil {
|
||||
t.Fatalf("create runtime session: %v", err)
|
||||
}
|
||||
stream := domain.LogStream{ID: "run.run-runtime.server-1.scum.console.stdout", ServerInstanceID: "server-1", Source: domain.LogStreamSourceProcess, StreamKey: "scum.console.stdout", LatestSeq: 42, StorageBackend: domain.LogStorageBackendLocalSegments, RetentionPolicy: "default", CreatedAt: stamp, UpdatedAt: stamp}
|
||||
if err := store.LogStreams().Create(stream); err != nil {
|
||||
t.Fatalf("create runtime log stream: %v", err)
|
||||
}
|
||||
mainAfter, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
t.Fatalf("read main snapshot after runtime update: %v", err)
|
||||
@@ -341,7 +345,7 @@ func TestFileStorePersistsRunRuntimeStateSeparately(t *testing.T) {
|
||||
if err != nil {
|
||||
t.Fatalf("read runtime snapshot: %v", err)
|
||||
}
|
||||
if !strings.Contains(string(runtimePayload), endpoint.ID) || strings.Contains(string(runtimePayload), session.SessionToken) {
|
||||
if !strings.Contains(string(runtimePayload), endpoint.ID) || !strings.Contains(string(runtimePayload), stream.ID) || strings.Contains(string(runtimePayload), session.SessionToken) {
|
||||
t.Fatalf("unexpected runtime snapshot payload: %s", runtimePayload)
|
||||
}
|
||||
|
||||
@@ -355,6 +359,9 @@ func TestFileStorePersistsRunRuntimeStateSeparately(t *testing.T) {
|
||||
if got, err := reloaded.RunControlSessions().Get(endpoint.ID); err != nil || got.SessionTokenHash != session.SessionTokenHash || got.UsedNonces[0] != "nonce-1" {
|
||||
t.Fatalf("runtime session did not reload: %+v err=%v", got, err)
|
||||
}
|
||||
if got, err := reloaded.LogStreams().Get(stream.ID); err != nil || got.LatestSeq != stream.LatestSeq {
|
||||
t.Fatalf("runtime log stream did not reload: %+v err=%v", got, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestFileStorePersistsPluginOperationsStateAcrossRestart(t *testing.T) {
|
||||
|
||||
Reference in New Issue
Block a user