mirror of
https://github.com/psviderski/uncloud.git
synced 2026-08-26 11:03:34 +00:00
fix: uncloud daemon crash when listing partially replicated container records
This commit is contained in:
@@ -82,7 +82,7 @@ func (s *Store) CreateOrUpdateContainer(ctx context.Context, ctr api.ServiceCont
|
||||
|
||||
// ListContainers returns a list of container records from the store database that match the given options.
|
||||
func (s *Store) ListContainers(ctx context.Context, opts ListOptions) ([]ContainerRecord, error) {
|
||||
q := sq.Select("container", "machine_id", "sync_status", "updated_at").From("containers").
|
||||
q := sq.Select("id", "container", "machine_id", "sync_status", "updated_at").From("containers").
|
||||
Where(sq.Eq{"sync_status": SyncStatusSynced})
|
||||
|
||||
if len(opts.MachineIDs) > 0 {
|
||||
@@ -112,14 +112,23 @@ func (s *Store) ListContainers(ctx context.Context, opts ListOptions) ([]Contain
|
||||
defer rows.Close()
|
||||
|
||||
var containers []ContainerRecord
|
||||
var cJSON, machineID, syncStatus, updatedAtStr string
|
||||
var id, cJSON, machineID, syncStatus, updatedAtStr string
|
||||
var updatedAt time.Time
|
||||
skipped := 0
|
||||
|
||||
for rows.Next() {
|
||||
if err = rows.Scan(&cJSON, &machineID, &syncStatus, &updatedAtStr); err != nil {
|
||||
if err = rows.Scan(&id, &cJSON, &machineID, &syncStatus, &updatedAtStr); err != nil {
|
||||
return nil, fmt.Errorf("scan container record: %w", err)
|
||||
}
|
||||
|
||||
// Skip containers with empty JSON data. This can happen during partial replication
|
||||
// when cr-sqlite has created the row but the container column hasn't been synced yet.
|
||||
if cJSON == "" || cJSON == "{}" {
|
||||
slog.Debug("Skipping container with empty data in the store (partial replication?).", "id", id)
|
||||
skipped++
|
||||
continue
|
||||
}
|
||||
|
||||
var c api.ServiceContainer
|
||||
if err = json.Unmarshal([]byte(cJSON), &c); err != nil {
|
||||
return nil, fmt.Errorf("unmarshal container: %w", err)
|
||||
@@ -135,6 +144,11 @@ func (s *Store) ListContainers(ctx context.Context, opts ListOptions) ([]Contain
|
||||
})
|
||||
}
|
||||
|
||||
if skipped > 0 {
|
||||
slog.Warn("Listing containers from the store skipped empty records (possibly due to partial replication).",
|
||||
"skipped", skipped, "valid", len(containers))
|
||||
}
|
||||
|
||||
return containers, nil
|
||||
}
|
||||
|
||||
@@ -166,7 +180,7 @@ func (s *Store) DeleteContainers(ctx context.Context, opts DeleteOptions) error
|
||||
// receive any values, it just signals when a container(s) has been added, updated, or deleted in the database.
|
||||
func (s *Store) SubscribeContainers(ctx context.Context) ([]ContainerRecord, <-chan struct{}, error) {
|
||||
// TODO: figure out whether we need sync_status at all (not used at the moment).
|
||||
q := sq.Select("container", "machine_id", "sync_status", "updated_at").From("containers").
|
||||
q := sq.Select("id", "container", "machine_id", "sync_status", "updated_at").From("containers").
|
||||
Where(sq.Eq{"sync_status": SyncStatusSynced})
|
||||
query, args, err := q.ToSql()
|
||||
if err != nil {
|
||||
@@ -179,15 +193,24 @@ func (s *Store) SubscribeContainers(ctx context.Context) ([]ContainerRecord, <-c
|
||||
}
|
||||
|
||||
var containers []ContainerRecord
|
||||
var cJSON, updatedAtStr string
|
||||
var id, cJSON, updatedAtStr string
|
||||
skipped := 0
|
||||
|
||||
rows := sub.Rows()
|
||||
for rows.Next() {
|
||||
var cr ContainerRecord
|
||||
if err = rows.Scan(&cJSON, &cr.MachineID, &cr.SyncStatus, &updatedAtStr); err != nil {
|
||||
if err = rows.Scan(&id, &cJSON, &cr.MachineID, &cr.SyncStatus, &updatedAtStr); err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
|
||||
// Skip containers with empty JSON data. This can happen during partial replication
|
||||
// when cr-sqlite has created the row but the container column hasn't been synced yet.
|
||||
if cJSON == "" || cJSON == "{}" {
|
||||
slog.Debug("Skipping container with empty data in the store (partial replication?).", "id", id)
|
||||
skipped++
|
||||
continue
|
||||
}
|
||||
|
||||
if err = json.Unmarshal([]byte(cJSON), &cr.Container); err != nil {
|
||||
return nil, nil, fmt.Errorf("unmarshal container: %w", err)
|
||||
}
|
||||
@@ -196,6 +219,12 @@ func (s *Store) SubscribeContainers(ctx context.Context) ([]ContainerRecord, <-c
|
||||
}
|
||||
containers = append(containers, cr)
|
||||
}
|
||||
|
||||
if skipped > 0 {
|
||||
slog.Warn("Container subscription skipped empty records in the store (possibly due to partial replication).",
|
||||
"skipped", skipped, "valid", len(containers))
|
||||
}
|
||||
|
||||
events, err := sub.Changes()
|
||||
if err != nil {
|
||||
return nil, nil, fmt.Errorf("get subscription changes: %w", err)
|
||||
|
||||
@@ -116,19 +116,29 @@ func (s *Store) GetMachine(ctx context.Context, machineID string) (*pb.MachineIn
|
||||
}
|
||||
|
||||
func (s *Store) ListMachines(ctx context.Context) ([]*pb.MachineInfo, error) {
|
||||
rows, err := s.corro.QueryContext(ctx, "SELECT info FROM machines ORDER BY name")
|
||||
rows, err := s.corro.QueryContext(ctx, "SELECT id, info FROM machines ORDER BY name")
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
var machines []*pb.MachineInfo
|
||||
skipped := 0
|
||||
|
||||
for rows.Next() {
|
||||
var mJSON string
|
||||
if err = rows.Scan(&mJSON); err != nil {
|
||||
var id, mJSON string
|
||||
if err = rows.Scan(&id, &mJSON); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// Skip machines with empty JSON data. This can happen during partial replication
|
||||
// when cr-sqlite has created the row but the info column hasn't been synced yet.
|
||||
if mJSON == "" || mJSON == "{}" {
|
||||
slog.Debug("Skipping machine with empty data in the store (partial replication?).", "id", id)
|
||||
skipped++
|
||||
continue
|
||||
}
|
||||
|
||||
protojsonParser := protojson.UnmarshalOptions{DiscardUnknown: true}
|
||||
var m pb.MachineInfo
|
||||
if err = protojsonParser.Unmarshal([]byte(mJSON), &m); err != nil {
|
||||
@@ -141,6 +151,12 @@ func (s *Store) ListMachines(ctx context.Context) ([]*pb.MachineInfo, error) {
|
||||
}
|
||||
machines = append(machines, &m)
|
||||
}
|
||||
|
||||
if skipped > 0 {
|
||||
slog.Warn("Listing machines from the store skipped empty records (possibly due to partial replication).",
|
||||
"skipped", skipped, "valid", len(machines))
|
||||
}
|
||||
|
||||
return machines, nil
|
||||
}
|
||||
|
||||
@@ -186,24 +202,41 @@ func (s *Store) DeleteMachine(ctx context.Context, id string) error {
|
||||
// SubscribeMachines returns a list of machines and a channel that signals changes to the list. The channel doesn't
|
||||
// receive any values, it just signals when a machine has been added, updated, or deleted in the database.
|
||||
func (s *Store) SubscribeMachines(ctx context.Context) ([]*pb.MachineInfo, <-chan struct{}, error) {
|
||||
sub, err := s.corro.SubscribeContext(ctx, "SELECT info FROM machines ORDER BY name", nil, false)
|
||||
sub, err := s.corro.SubscribeContext(ctx, "SELECT id, info FROM machines ORDER BY name", nil, false)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
|
||||
rows := sub.Rows()
|
||||
var machines []*pb.MachineInfo
|
||||
skipped := 0
|
||||
|
||||
for rows.Next() {
|
||||
var mJSON string
|
||||
if err = rows.Scan(&mJSON); err != nil {
|
||||
var id, mJSON string
|
||||
if err = rows.Scan(&id, &mJSON); err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
|
||||
// Skip machines with empty JSON data. This can happen during partial replication
|
||||
// when cr-sqlite has created the row but the info column hasn't been synced yet.
|
||||
if mJSON == "" || mJSON == "{}" {
|
||||
slog.Debug("Skipping machine with empty data in the store (partial replication?).", "id", id)
|
||||
skipped++
|
||||
continue
|
||||
}
|
||||
|
||||
var m pb.MachineInfo
|
||||
if err = protojson.Unmarshal([]byte(mJSON), &m); err != nil {
|
||||
return nil, nil, fmt.Errorf("unmarshal machine info: %w", err)
|
||||
}
|
||||
machines = append(machines, &m)
|
||||
}
|
||||
|
||||
if skipped > 0 {
|
||||
slog.Warn("Machine subscription skipped empty records in the store (possibly due to partial replication).",
|
||||
"skipped", skipped, "valid", len(machines))
|
||||
}
|
||||
|
||||
events, err := sub.Changes()
|
||||
if err != nil {
|
||||
return nil, nil, fmt.Errorf("get subscription changes: %w", err)
|
||||
|
||||
@@ -135,6 +135,10 @@ func (c *Container) UnmarshalJSON(data []byte) error {
|
||||
}
|
||||
|
||||
*c = Container(temp)
|
||||
if c.ContainerJSONBase == nil {
|
||||
return fmt.Errorf("container data is missing mandatory base fields: %s", data)
|
||||
}
|
||||
|
||||
c.Name = strings.TrimPrefix(c.Name, "/")
|
||||
|
||||
return nil
|
||||
|
||||
Reference in New Issue
Block a user