RomanNum3ral
/
Cybersecurity-Projects
mirror of https://github.com/CarterPerez-dev/Cybersecurity-Projects.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

feat(monitor/collectors/state): collector_state CRUD with healthy/degraded upserts

This commit is contained in:
CarterPerez-dev 2026-05-01 22:26:57 -04:00
parent c5cc7951a3
commit 91d855f4ef
4 changed files with 207 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
PROJECTS/advanced/monitor-the-situation-dashboard/backend/go.mod
View File

@ -92,6 +92,7 @@ require (
github.com/shirou/gopsutil/v4 v4.26.3 // indirect
github.com/sirupsen/logrus v1.9.4 // indirect
github.com/testcontainers/testcontainers-go v0.42.0 // indirect
github.com/testcontainers/testcontainers-go/modules/postgres v0.42.0 // indirect
github.com/tklauser/go-sysconf v0.3.16 // indirect
github.com/tklauser/numcpus v0.11.0 // indirect
github.com/valyala/fastjson v1.6.4 // indirect

2
PROJECTS/advanced/monitor-the-situation-dashboard/backend/go.sum
View File

@ -192,6 +192,8 @@ github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu
github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
github.com/testcontainers/testcontainers-go v0.42.0 h1:He3IhTzTZOygSXLJPMX7n44XtK+qhjat1nI9cneBbUY=
github.com/testcontainers/testcontainers-go v0.42.0/go.mod h1:vZjdY1YmUA1qEForxOIOazfsrdyORJAbhi0bp8plN30=
github.com/testcontainers/testcontainers-go/modules/postgres v0.42.0 h1:GCbb1ndrF7OTDiIvxXyItaDab4qkzTFJ48LKFdM7EIo=
github.com/testcontainers/testcontainers-go/modules/postgres v0.42.0/go.mod h1:IRPBaI8jXdrNfD0e4Zm7Fbcgaz5shKxOQv4axiL09xs=
github.com/testcontainers/testcontainers-go/modules/redis v0.42.0 h1:id/6LH8ZeDrtAUVSuNvZUAJ1kVpb82y1pr9yweAWsRg=
github.com/testcontainers/testcontainers-go/modules/redis v0.42.0/go.mod h1:uF0jI8FITagQpBNOgweGBmPf6rP4K0SeL1XFPbsZSSY=
github.com/tklauser/go-sysconf v0.3.16 h1:frioLaCQSsF5Cy1jgRBrzr6t502KIIwQ0MArYICU0nA=

94
PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/state/repo.go Normal file
View File

@ -0,0 +1,94 @@
// ©AngelaMos | 2026
// repo.go
package state
import (
"context"
"fmt"
"time"
"github.com/jmoiron/sqlx"
)
type CollectorState string
const (
StateHealthy CollectorState = "healthy"
StateDegraded CollectorState = "degraded"
StateDown CollectorState = "down"
)
type Row struct {
Name string `db:"name"`
State CollectorState `db:"state"`
LastSuccessAt *time.Time `db:"last_success_at"`
LastErrorAt *time.Time `db:"last_error_at"`
LastError string `db:"last_error"`
LastEventCount int64 `db:"last_event_count"`
UpdatedAt time.Time `db:"updated_at"`
}
type Repo struct {
db *sqlx.DB
}
func NewRepo(db *sqlx.DB) *Repo { return &Repo{db: db} }
func (r *Repo) RecordSuccess(ctx context.Context, name string, eventCount int64) error {
_, err := r.db.ExecContext(ctx, `
INSERT INTO collector_state (name, state, last_success_at, last_event_count, updated_at)
VALUES ($1, $2, now(), $3, now())
ON CONFLICT (name) DO UPDATE SET
state = EXCLUDED.state,
last_success_at = EXCLUDED.last_success_at,
last_event_count = collector_state.last_event_count + EXCLUDED.last_event_count,
updated_at = EXCLUDED.updated_at`,
name, StateHealthy, eventCount,
)
if err != nil {
return fmt.Errorf("upsert healthy %s: %w", name, err)
}
return nil
}
func (r *Repo) RecordError(ctx context.Context, name, errMsg string) error {
_, err := r.db.ExecContext(ctx, `
INSERT INTO collector_state (name, state, last_error_at, last_error, updated_at)
VALUES ($1, $2, now(), $3, now())
ON CONFLICT (name) DO UPDATE SET
state = EXCLUDED.state,
last_error_at = EXCLUDED.last_error_at,
last_error = EXCLUDED.last_error,
updated_at = EXCLUDED.updated_at`,
name, StateDegraded, errMsg,
)
if err != nil {
return fmt.Errorf("upsert degraded %s: %w", name, err)
}
return nil
}
func (r *Repo) Get(ctx context.Context, name string) (Row, error) {
var row Row
err := r.db.GetContext(ctx, &row, `
SELECT name, state, last_success_at, last_error_at, last_error,
last_event_count, updated_at
FROM collector_state WHERE name = $1`, name)
if err != nil {
return Row{}, fmt.Errorf("get state %s: %w", name, err)
}
return row, nil
}
func (r *Repo) All(ctx context.Context) ([]Row, error) {
var rows []Row
err := r.db.SelectContext(ctx, &rows, `
SELECT name, state, last_success_at, last_error_at, last_error,
last_event_count, updated_at
FROM collector_state ORDER BY name`)
if err != nil {
return nil, fmt.Errorf("select all collector_state: %w", err)
}
return rows, nil
}

110
PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/state/repo_test.go Normal file
View File

@ -0,0 +1,110 @@
// ©AngelaMos | 2026
// repo_test.go
package state_test
import (
"context"
"testing"
"time"
_ "github.com/jackc/pgx/v5/stdlib"
"github.com/jmoiron/sqlx"
"github.com/stretchr/testify/require"
"github.com/testcontainers/testcontainers-go/modules/postgres"
"github.com/carterperez-dev/monitor-the-situation/backend/internal/collectors/state"
)
func setupDB(t *testing.T) *sqlx.DB {
t.Helper()
ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second)
defer cancel()
pg, err := postgres.Run(ctx,
"postgres:17-alpine",
postgres.WithDatabase("monitor"),
postgres.WithUsername("monitor"),
postgres.WithPassword("monitor"),
postgres.BasicWaitStrategies(),
)
require.NoError(t, err)
t.Cleanup(func() { _ = pg.Terminate(context.Background()) })
dsn, err := pg.ConnectionString(ctx, "sslmode=disable")
require.NoError(t, err)
db, err := sqlx.ConnectContext(ctx, "pgx", dsn)
require.NoError(t, err)
t.Cleanup(func() { _ = db.Close() })
_, err = db.ExecContext(ctx, `
CREATE TABLE collector_state (
name text PRIMARY KEY,
state text NOT NULL,
last_success_at timestamptz,
last_error_at timestamptz,
last_error text NOT NULL DEFAULT '',
last_event_count bigint NOT NULL DEFAULT 0,
updated_at timestamptz NOT NULL DEFAULT now()
)`)
require.NoError(t, err)
return db
}
func TestRepo_RecordSuccessThenError(t *testing.T) {
db := setupDB(t)
repo := state.NewRepo(db)
ctx := context.Background()
require.NoError(t, repo.RecordSuccess(ctx, "dshield", 12))
got, err := repo.Get(ctx, "dshield")
require.NoError(t, err)
require.Equal(t, "dshield", got.Name)
require.Equal(t, state.StateHealthy, got.State)
require.EqualValues(t, 12, got.LastEventCount)
require.NotNil(t, got.LastSuccessAt)
require.WithinDuration(t, time.Now(), *got.LastSuccessAt, 5*time.Second)
require.NoError(t, repo.RecordError(ctx, "dshield", "upstream 503"))
got, err = repo.Get(ctx, "dshield")
require.NoError(t, err)
require.Equal(t, state.StateDegraded, got.State)
require.Equal(t, "upstream 503", got.LastError)
require.NotNil(t, got.LastErrorAt)
require.WithinDuration(t, time.Now(), *got.LastErrorAt, 5*time.Second)
}
func TestRepo_SuccessAccumulatesCount(t *testing.T) {
db := setupDB(t)
repo := state.NewRepo(db)
ctx := context.Background()
require.NoError(t, repo.RecordSuccess(ctx, "kev", 3))
require.NoError(t, repo.RecordSuccess(ctx, "kev", 5))
got, err := repo.Get(ctx, "kev")
require.NoError(t, err)
require.EqualValues(t, 8, got.LastEventCount)
}
func TestRepo_AllReturnsRowsSorted(t *testing.T) {
db := setupDB(t)
repo := state.NewRepo(db)
ctx := context.Background()
require.NoError(t, repo.RecordSuccess(ctx, "ransomware", 1))
require.NoError(t, repo.RecordSuccess(ctx, "cve", 1))
require.NoError(t, repo.RecordSuccess(ctx, "kev", 1))
rows, err := repo.All(ctx)
require.NoError(t, err)
require.Len(t, rows, 3)
require.Equal(t, "cve", rows[0].Name)
require.Equal(t, "kev", rows[1].Name)
require.Equal(t, "ransomware", rows[2].Name)
}