openstatus.dev
Openstatus
www.openstatus.dev
private-location-update: add tests (#1766)
* improve private-location :
* some change
* add tcp monitor
* add dns monitor
* add tests
authored by
Thibault Le Ouay
and committed by
GitHub
9a5ee947
343ee0cd
+7
-7
apps/checker/proto/private_location/v1/assertions.pb.go
+7
-7
apps/checker/proto/private_location/v1/assertions.pb.go
···
1
1
// Code generated by protoc-gen-go. DO NOT EDIT.
2
2
// versions:
3
-
// protoc-gen-go v1.36.10
3
+
// protoc-gen-go v1.36.11
4
4
// protoc (unknown)
5
5
// source: private_location/v1/assertions.proto
6
6
···
378
378
state protoimpl.MessageState `protogen:"open.v1"`
379
379
Record string `protobuf:"bytes,1,opt,name=record,proto3" json:"record,omitempty"`
380
380
Comparator RecordComparator `protobuf:"varint,2,opt,name=comparator,proto3,enum=private_location.v1.RecordComparator" json:"comparator,omitempty"`
381
-
Targert string `protobuf:"bytes,3,opt,name=targert,proto3" json:"targert,omitempty"`
381
+
Target string `protobuf:"bytes,3,opt,name=target,proto3" json:"target,omitempty"`
382
382
unknownFields protoimpl.UnknownFields
383
383
sizeCache protoimpl.SizeCache
384
384
}
···
427
427
return RecordComparator_RECORD_COMPARATOR_UNSPECIFIED
428
428
}
429
429
430
-
func (x *RecordAssertion) GetTargert() string {
430
+
func (x *RecordAssertion) GetTarget() string {
431
431
if x != nil {
432
-
return x.Targert
432
+
return x.Target
433
433
}
434
434
return ""
435
435
}
···
454
454
"\n" +
455
455
"comparator\x18\x02 \x01(\x0e2%.private_location.v1.StringComparatorR\n" +
456
456
"comparator\x12\x10\n" +
457
-
"\x03key\x18\x03 \x01(\tR\x03key\"\x8a\x01\n" +
457
+
"\x03key\x18\x03 \x01(\tR\x03key\"\x88\x01\n" +
458
458
"\x0fRecordAssertion\x12\x16\n" +
459
459
"\x06record\x18\x01 \x01(\tR\x06record\x12E\n" +
460
460
"\n" +
461
461
"comparator\x18\x02 \x01(\x0e2%.private_location.v1.RecordComparatorR\n" +
462
-
"comparator\x12\x18\n" +
463
-
"\atargert\x18\x03 \x01(\tR\atargert*\x8f\x02\n" +
462
+
"comparator\x12\x16\n" +
463
+
"\x06target\x18\x03 \x01(\tR\x06target*\x8f\x02\n" +
464
464
"\x10NumberComparator\x12!\n" +
465
465
"\x1dNUMBER_COMPARATOR_UNSPECIFIED\x10\x00\x12\x1b\n" +
466
466
"\x17NUMBER_COMPARATOR_EQUAL\x10\x01\x12\x1f\n" +
+1
-1
apps/checker/proto/private_location/v1/dns_monitor.pb.go
+1
-1
apps/checker/proto/private_location/v1/dns_monitor.pb.go
+1
-1
apps/checker/proto/private_location/v1/http_monitor.pb.go
+1
-1
apps/checker/proto/private_location/v1/http_monitor.pb.go
+1
-1
apps/checker/proto/private_location/v1/private_location.pb.go
+1
-1
apps/checker/proto/private_location/v1/private_location.pb.go
+1
-1
apps/checker/proto/private_location/v1/tcp_monitor.pb.go
+1
-1
apps/checker/proto/private_location/v1/tcp_monitor.pb.go
+12
-6
apps/private-location/internal/database/database.go
+12
-6
apps/private-location/internal/database/database.go
···
1
1
package database
2
2
3
3
import (
4
-
// "database/sql"
5
4
"database/sql"
6
5
"fmt"
7
-
8
-
// "log"
9
6
"os"
10
7
11
8
"github.com/jmoiron/sqlx"
12
9
_ "github.com/joho/godotenv/autoload"
13
-
// "github.com/tursodatabase/go-libsql"
14
10
_ "github.com/tursodatabase/libsql-client-go/libsql"
15
11
)
16
12
17
-
var DB *sqlx.DB
18
-
19
13
var (
20
14
dbUrl = os.Getenv("DB_URL")
21
15
authToken = os.Getenv("DB_AUTH_TOKEN")
22
16
dbInstance *sqlx.DB
23
17
)
24
18
19
+
// New returns a database connection, reusing an existing connection if available.
25
20
func New() *sqlx.DB {
26
21
// Reuse Connection
27
22
if dbInstance != nil {
···
36
31
}
37
32
38
33
db := sqlx.NewDb(c, "sqlite3")
34
+
dbInstance = db
39
35
40
36
return db
41
37
}
38
+
39
+
// Close closes the database connection.
40
+
func Close() error {
41
+
if dbInstance != nil {
42
+
err := dbInstance.Close()
43
+
dbInstance = nil
44
+
return err
45
+
}
46
+
return nil
47
+
}
+323
apps/private-location/internal/logs/logs_test.go
+323
apps/private-location/internal/logs/logs_test.go
···
1
+
package logs_test
2
+
3
+
import (
4
+
"log/slog"
5
+
"testing"
6
+
"time"
7
+
8
+
"github.com/openstatushq/openstatus/apps/private-location/internal/logs"
9
+
)
10
+
11
+
func TestShouldSample(t *testing.T) {
12
+
tests := []struct {
13
+
name string
14
+
event map[string]any
15
+
expected bool
16
+
}{
17
+
{
18
+
name: "server error 500 should always sample",
19
+
event: map[string]any{
20
+
"status_code": 500,
21
+
},
22
+
expected: true,
23
+
},
24
+
{
25
+
name: "server error 503 should always sample",
26
+
event: map[string]any{
27
+
"status_code": 503,
28
+
},
29
+
expected: true,
30
+
},
31
+
{
32
+
name: "server error 599 should always sample",
33
+
event: map[string]any{
34
+
"status_code": 599,
35
+
},
36
+
expected: true,
37
+
},
38
+
{
39
+
name: "explicit error should always sample",
40
+
event: map[string]any{
41
+
"status_code": 200,
42
+
"error": "something went wrong",
43
+
},
44
+
expected: true,
45
+
},
46
+
{
47
+
name: "slow request above 2000ms should always sample",
48
+
event: map[string]any{
49
+
"status_code": 200,
50
+
"duration_ms": 2001,
51
+
},
52
+
expected: true,
53
+
},
54
+
{
55
+
name: "slow request exactly 2000ms should not always sample",
56
+
event: map[string]any{
57
+
"status_code": 200,
58
+
"duration_ms": 2000,
59
+
},
60
+
expected: false, // This will be randomly sampled at 20%
61
+
},
62
+
{
63
+
name: "client error 400 should always sample",
64
+
event: map[string]any{
65
+
"status_code": 400,
66
+
},
67
+
expected: true,
68
+
},
69
+
{
70
+
name: "client error 404 should always sample",
71
+
event: map[string]any{
72
+
"status_code": 404,
73
+
},
74
+
expected: true,
75
+
},
76
+
{
77
+
name: "client error 499 should always sample",
78
+
event: map[string]any{
79
+
"status_code": 499,
80
+
},
81
+
expected: true,
82
+
},
83
+
{
84
+
name: "status code 399 should not always sample (below client error range)",
85
+
event: map[string]any{
86
+
"status_code": 399,
87
+
},
88
+
expected: false, // Random 20% sampling
89
+
},
90
+
}
91
+
92
+
for _, tt := range tests {
93
+
t.Run(tt.name, func(t *testing.T) {
94
+
result := logs.ShouldSample(tt.event)
95
+
if tt.expected && !result {
96
+
t.Errorf("ShouldSample() = %v, expected %v (should always sample)", result, tt.expected)
97
+
}
98
+
// For cases where expected is false, we can't deterministically test
99
+
// because the function uses random sampling. We just verify it doesn't
100
+
// always return true.
101
+
})
102
+
}
103
+
}
104
+
105
+
func TestShouldSample_RandomSampling(t *testing.T) {
106
+
// Test that successful, fast requests are sometimes sampled
107
+
event := map[string]any{
108
+
"status_code": 200,
109
+
"duration_ms": 100,
110
+
}
111
+
112
+
// Run multiple times to verify random sampling works
113
+
sampledCount := 0
114
+
iterations := 1000
115
+
for i := 0; i < iterations; i++ {
116
+
if logs.ShouldSample(event) {
117
+
sampledCount++
118
+
}
119
+
}
120
+
121
+
// With 20% sampling, we expect roughly 200 samples out of 1000
122
+
// Allow for some variance (between 10% and 30%)
123
+
minExpected := iterations / 10 // 10%
124
+
maxExpected := iterations * 3 / 10 // 30%
125
+
126
+
if sampledCount < minExpected || sampledCount > maxExpected {
127
+
t.Errorf("Random sampling seems off: got %d samples out of %d (expected roughly 20%%)", sampledCount, iterations)
128
+
}
129
+
}
130
+
131
+
func TestShouldSample_EmptyEvent(t *testing.T) {
132
+
// Empty event should fall through to random sampling
133
+
event := map[string]any{}
134
+
135
+
// Just verify it doesn't panic
136
+
_ = logs.ShouldSample(event)
137
+
}
138
+
139
+
func TestShouldSample_MissingFields(t *testing.T) {
140
+
// Event with no status_code or duration_ms
141
+
event := map[string]any{
142
+
"path": "/api/test",
143
+
"method": "GET",
144
+
}
145
+
146
+
// Should fall through to random sampling without panic
147
+
_ = logs.ShouldSample(event)
148
+
}
149
+
150
+
func TestMapToAttrs(t *testing.T) {
151
+
tests := []struct {
152
+
name string
153
+
input map[string]any
154
+
expected int // expected number of attributes
155
+
}{
156
+
{
157
+
name: "empty map",
158
+
input: map[string]any{},
159
+
expected: 0,
160
+
},
161
+
{
162
+
name: "single string value",
163
+
input: map[string]any{
164
+
"key": "value",
165
+
},
166
+
expected: 1,
167
+
},
168
+
{
169
+
name: "multiple types",
170
+
input: map[string]any{
171
+
"string_key": "value",
172
+
"int_key": 42,
173
+
"bool_key": true,
174
+
},
175
+
expected: 3,
176
+
},
177
+
}
178
+
179
+
for _, tt := range tests {
180
+
t.Run(tt.name, func(t *testing.T) {
181
+
attrs := logs.MapToAttrs(tt.input)
182
+
if len(attrs) != tt.expected {
183
+
t.Errorf("MapToAttrs() returned %d attrs, expected %d", len(attrs), tt.expected)
184
+
}
185
+
})
186
+
}
187
+
}
188
+
189
+
func TestMapToAttrs_TypeConversions(t *testing.T) {
190
+
now := time.Now()
191
+
duration := 5 * time.Second
192
+
193
+
input := map[string]any{
194
+
"string_val": "hello",
195
+
"int_val": 42,
196
+
"int64_val": int64(1234567890),
197
+
"float64_val": 3.14,
198
+
"bool_val": true,
199
+
"time_val": now,
200
+
"duration_val": duration,
201
+
}
202
+
203
+
attrs := logs.MapToAttrs(input)
204
+
205
+
// Verify correct number of attributes
206
+
if len(attrs) != 7 {
207
+
t.Errorf("Expected 7 attributes, got %d", len(attrs))
208
+
}
209
+
210
+
// Verify each attribute type
211
+
attrMap := make(map[string]slog.Attr)
212
+
for _, attr := range attrs {
213
+
attrMap[attr.Key] = attr
214
+
}
215
+
216
+
// Check string
217
+
if attr, ok := attrMap["string_val"]; ok {
218
+
if attr.Value.Kind() != slog.KindString {
219
+
t.Errorf("string_val should be String kind, got %v", attr.Value.Kind())
220
+
}
221
+
if attr.Value.String() != "hello" {
222
+
t.Errorf("string_val should be 'hello', got %v", attr.Value.String())
223
+
}
224
+
}
225
+
226
+
// Check int
227
+
if attr, ok := attrMap["int_val"]; ok {
228
+
if attr.Value.Kind() != slog.KindInt64 {
229
+
t.Errorf("int_val should be Int64 kind, got %v", attr.Value.Kind())
230
+
}
231
+
if attr.Value.Int64() != 42 {
232
+
t.Errorf("int_val should be 42, got %v", attr.Value.Int64())
233
+
}
234
+
}
235
+
236
+
// Check bool
237
+
if attr, ok := attrMap["bool_val"]; ok {
238
+
if attr.Value.Kind() != slog.KindBool {
239
+
t.Errorf("bool_val should be Bool kind, got %v", attr.Value.Kind())
240
+
}
241
+
if attr.Value.Bool() != true {
242
+
t.Errorf("bool_val should be true, got %v", attr.Value.Bool())
243
+
}
244
+
}
245
+
}
246
+
247
+
func TestMapToAttrs_NestedMap(t *testing.T) {
248
+
input := map[string]any{
249
+
"outer": map[string]any{
250
+
"inner_string": "nested_value",
251
+
"inner_int": 123,
252
+
},
253
+
}
254
+
255
+
attrs := logs.MapToAttrs(input)
256
+
257
+
if len(attrs) != 1 {
258
+
t.Errorf("Expected 1 attribute (group), got %d", len(attrs))
259
+
}
260
+
261
+
// The nested map should be converted to a Group
262
+
if attrs[0].Key != "outer" {
263
+
t.Errorf("Expected key 'outer', got %s", attrs[0].Key)
264
+
}
265
+
if attrs[0].Value.Kind() != slog.KindGroup {
266
+
t.Errorf("Expected Group kind for nested map, got %v", attrs[0].Value.Kind())
267
+
}
268
+
}
269
+
270
+
func TestMapToAttrs_UnknownType(t *testing.T) {
271
+
type customType struct {
272
+
Field string
273
+
}
274
+
275
+
input := map[string]any{
276
+
"custom": customType{Field: "test"},
277
+
}
278
+
279
+
attrs := logs.MapToAttrs(input)
280
+
281
+
if len(attrs) != 1 {
282
+
t.Errorf("Expected 1 attribute, got %d", len(attrs))
283
+
}
284
+
285
+
// Unknown types should be converted using slog.Any
286
+
if attrs[0].Key != "custom" {
287
+
t.Errorf("Expected key 'custom', got %s", attrs[0].Key)
288
+
}
289
+
if attrs[0].Value.Kind() != slog.KindAny {
290
+
t.Errorf("Expected Any kind for unknown type, got %v", attrs[0].Value.Kind())
291
+
}
292
+
}
293
+
294
+
func TestMapToAny(t *testing.T) {
295
+
input := map[string]any{
296
+
"key1": "value1",
297
+
"key2": 42,
298
+
}
299
+
300
+
result := logs.MapToAny(input)
301
+
302
+
// MapToAny returns []any containing slog.Attr values
303
+
if len(result) != 2 {
304
+
t.Errorf("Expected 2 items, got %d", len(result))
305
+
}
306
+
307
+
// Verify each item is an slog.Attr
308
+
for _, item := range result {
309
+
if _, ok := item.(slog.Attr); !ok {
310
+
t.Errorf("Expected slog.Attr, got %T", item)
311
+
}
312
+
}
313
+
}
314
+
315
+
func TestMapToAny_EmptyMap(t *testing.T) {
316
+
input := map[string]any{}
317
+
318
+
result := logs.MapToAny(input)
319
+
320
+
if len(result) != 0 {
321
+
t.Errorf("Expected empty slice, got %d items", len(result))
322
+
}
323
+
}
+21
-4
apps/private-location/internal/models/assertions.go
+21
-4
apps/private-location/internal/models/assertions.go
···
5
5
type AssertionType string
6
6
7
7
const (
8
-
AssertionHeader AssertionType = "header"
9
-
AssertionTextBody AssertionType = "textBody"
10
-
AssertionStatus AssertionType = "status"
11
-
AssertionJsonBody AssertionType = "jsonBody"
8
+
AssertionHeader AssertionType = "header"
9
+
AssertionTextBody AssertionType = "textBody"
10
+
AssertionStatus AssertionType = "status"
11
+
AssertionJsonBody AssertionType = "jsonBody"
12
+
AssertionDnsRecord AssertionType = "dnsRecord"
12
13
)
13
14
14
15
type StringComparator string
···
74
75
Comparator StringComparator `json:"compare"`
75
76
Target string `json:"target"`
76
77
}
78
+
79
+
type RecordComparator string
80
+
81
+
const (
82
+
RecordEquals RecordComparator = "eq"
83
+
RecordNotEquals RecordComparator = "not_eq"
84
+
RecordContains RecordComparator = "contains"
85
+
RecordNotContains RecordComparator = "not_contains"
86
+
)
87
+
88
+
type RecordTarget struct {
89
+
AssertionType AssertionType `json:"type"`
90
+
Comparator RecordComparator `json:"compare"`
91
+
Target string `json:"target"`
92
+
Key string `json:"key"`
93
+
}
+6
-2
apps/private-location/internal/server/db_testdata
+6
-2
apps/private-location/internal/server/db_testdata
···
436
436
('2', 'http', '10m', '0', 'https://www.google.com', '', '', '1', '', '', 'GET', '1760358329', 'gru', '1760358329', 'active', NULL, NULL, '1', '45000', NULL, NULL, NULL, '3', '1'),
437
437
('3', 'http', '1m', '1', 'https://www.openstatus.dev', 'OpenStatus', 'OpenStatus website', '1', '[{"key":"key", "value":"value"}]', '{"hello":"world"}', 'GET', '1760358329', 'ams', '1760358329', 'active', NULL, NULL, '0', '45000', NULL, NULL, NULL, '3', '1'),
438
438
('4', 'http', '10m', '1', 'https://www.google.com', '', '', '1', '', '', 'GET', '1760358329', 'gru', '1760358329', 'active', NULL, NULL, '1', '45000', NULL, 'https://otel.com:4337', '[{"key":"Authorization","value":"Basic"}]', '3', '1'),
439
-
('5', 'http', '10m', '1', 'https://openstat.us', '', '', '3', '', '', 'GET', '1760358329', 'ams', '1760358329', 'active', NULL, NULL, '1', '45000', NULL, NULL, NULL, '3', '1');
439
+
('5', 'http', '10m', '1', 'https://openstat.us', '', '', '3', '', '', 'GET', '1760358329', 'ams', '1760358329', 'active', NULL, NULL, '1', '45000', NULL, NULL, NULL, '3', '1'),
440
+
('6', 'tcp', '5m', '1', 'tcp://db.example.com:5432', 'Database TCP', 'Database TCP check', '3', '', '', '', '1760358329', 'ams', '1760358329', 'active', NULL, NULL, '0', '30000', '5000', NULL, NULL, '2', '0'),
441
+
('7', 'dns', '5m', '1', 'openstatus.dev', 'DNS Check', 'DNS check for openstatus.dev', '3', '', '', '', '1760358329', 'ams', '1760358329', 'active', '[{"version":"v1","type":"dnsRecord","key":"A","compare":"contains","target":"76.76.21.21"}]', NULL, '0', '30000', '3000', NULL, NULL, '2', '0');
440
442
441
443
INSERT INTO "notification" ("id", "name", "provider", "data", "workspace_id", "created_at", "updated_at") VALUES
442
444
('1', 'sample test notification', 'email', '{"email":"ping@openstatus.dev"}', '1', '1760358329', '1760358329');
···
462
464
('1', 'My Home', 'my-secret-key', NULL, '3', '1760358329', '1760358329');
463
465
464
466
INSERT INTO "private_location_to_monitor" ("private_location_id", "monitor_id", "created_at", "deleted_at") VALUES
465
-
('1', '5', '1760358329', NULL);
467
+
('1', '5', '1760358329', NULL),
468
+
('1', '6', '1760358329', NULL),
469
+
('1', '7', '1760358329', NULL);
+10
apps/private-location/internal/server/errors.go
+10
apps/private-location/internal/server/errors.go
···
1
+
package server
2
+
3
+
import "errors"
4
+
5
+
// Common errors used across the server package
6
+
var (
7
+
ErrMissingToken = errors.New("missing token")
8
+
ErrMonitorNotFound = errors.New("monitor not found")
9
+
ErrPrivateLocationNotFound = errors.New("private location not found")
10
+
)
+52
apps/private-location/internal/server/ingest_common.go
+52
apps/private-location/internal/server/ingest_common.go
···
1
+
package server
2
+
3
+
import (
4
+
"context"
5
+
"time"
6
+
7
+
"github.com/openstatushq/openstatus/apps/private-location/internal/database"
8
+
"github.com/rs/zerolog/log"
9
+
)
10
+
11
+
// ingestContext holds common data needed for ingestion
12
+
type ingestContext struct {
13
+
Monitor database.Monitor
14
+
Region database.PrivateLocation
15
+
}
16
+
17
+
// getIngestContext retrieves monitor and private location data for ingestion
18
+
func (h *privateLocationHandler) getIngestContext(ctx context.Context, token string, monitorID string) (*ingestContext, error) {
19
+
var monitor database.Monitor
20
+
err := h.db.Get(&monitor, "SELECT monitor.id, monitor.workspace_id, monitor.url, monitor.method, monitor.assertions FROM monitor JOIN private_location_to_monitor a ON monitor.id = a.monitor_id JOIN private_location b ON a.private_location_id = b.id WHERE b.token = ? AND monitor.deleted_at IS NULL and monitor.id = ?", token, monitorID)
21
+
if err != nil {
22
+
log.Ctx(ctx).Error().Err(err).Msg("failed to get monitor")
23
+
return nil, err
24
+
}
25
+
26
+
var region database.PrivateLocation
27
+
err = h.db.Get(®ion, "SELECT private_location.id FROM private_location join private_location_to_monitor a ON private_location.id = a.private_location_id WHERE a.monitor_id = ? AND private_location.token = ?", monitor.ID, token)
28
+
if err != nil {
29
+
log.Ctx(ctx).Error().Err(err).Msg("failed to get private location")
30
+
return nil, err
31
+
}
32
+
33
+
return &ingestContext{
34
+
Monitor: monitor,
35
+
Region: region,
36
+
}, nil
37
+
}
38
+
39
+
// sendEventAndUpdateLastSeen sends the event to Tinybird and updates the last_seen_at timestamp
40
+
func (h *privateLocationHandler) sendEventAndUpdateLastSeen(ctx context.Context, data any, dataSourceName string, regionID int) {
41
+
if err := h.TbClient.SendEvent(ctx, data, dataSourceName); err != nil {
42
+
log.Ctx(ctx).Error().Err(err).Msg("failed to send event to tinybird")
43
+
}
44
+
45
+
_, err := h.db.NamedExec("UPDATE private_location SET last_seen_at = :last_seen_at WHERE id = :id", map[string]any{
46
+
"last_seen_at": time.Now().Unix(),
47
+
"id": regionID,
48
+
})
49
+
if err != nil {
50
+
log.Ctx(ctx).Error().Err(err).Msg("failed to update private location")
51
+
}
52
+
}
+23
-33
apps/private-location/internal/server/ingest_dns.go
+23
-33
apps/private-location/internal/server/ingest_dns.go
···
2
2
3
3
import (
4
4
"context"
5
-
"errors"
6
5
"strconv"
7
-
"time"
8
6
9
7
"connectrpc.com/connect"
10
-
"github.com/openstatushq/openstatus/apps/private-location/internal/database"
8
+
"github.com/openstatushq/openstatus/apps/private-location/internal/tinybird"
11
9
private_locationv1 "github.com/openstatushq/openstatus/apps/private-location/proto/private_location/v1"
12
-
"github.com/rs/zerolog/log"
13
10
)
14
11
15
12
type DNSResponse struct {
···
35
32
func (h *privateLocationHandler) IngestDNS(ctx context.Context, req *connect.Request[private_locationv1.IngestDNSRequest]) (*connect.Response[private_locationv1.IngestDNSResponse], error) {
36
33
token := req.Header().Get("openstatus-token")
37
34
if token == "" {
38
-
return nil, connect.NewError(connect.CodeUnauthenticated, errors.New("missing token"))
35
+
return nil, connect.NewError(connect.CodeUnauthenticated, ErrMissingToken)
39
36
}
40
37
41
-
dataSourceName := "tcp_dns__v0"
42
-
43
-
var monitors database.Monitor
44
-
err := h.db.Get(&monitors, "SELECT monitor.* FROM monitor JOIN private_location_to_monitor a ON monitor.id = a.monitor_id JOIN private_location b ON a.private_location_id = b.id WHERE b.token = ? AND monitor.deleted_at IS NULL and monitor.id = ?", token, req.Msg.Id)
38
+
if err := ValidateIngestDNSRequest(req.Msg); err != nil {
39
+
return nil, NewValidationError(err)
40
+
}
45
41
42
+
ic, err := h.getIngestContext(ctx, token, req.Msg.Id)
46
43
if err != nil {
47
44
return nil, connect.NewError(connect.CodeInternal, err)
48
45
}
49
46
50
-
var region database.PrivateLocation
51
-
err = h.db.Get(®ion, "SELECT private_location.id FROM private_location join private_location_to_monitor a ON private_location.id = a.private_location_id WHERE a.monitor_id = ? and private_location.token = ?", monitors.ID, token)
52
-
53
-
if err != nil {
54
-
return nil, connect.NewError(connect.CodeInternal, err)
47
+
event := ctx.Value("event")
48
+
if eventMap, ok := event.(map[string]any); ok && eventMap != nil {
49
+
eventMap["private_location"] = map[string]any{
50
+
"monitor_id": req.Msg.MonitorId,
51
+
}
52
+
ctx = context.WithValue(ctx, "event", eventMap)
55
53
}
54
+
56
55
records := make(map[string][]string)
57
56
for _, record := range req.Msg.Records {
58
57
r := []string{}
···
63
62
}
64
63
65
64
data := DNSResponse{
66
-
ID: req.Msg.Id,
67
-
WorkspaceID: int64(monitors.WorkspaceID),
68
-
Timestamp: req.Msg.Timestamp,
69
-
Error: uint8(req.Msg.Error),
70
-
// ErrorMessage: req.Msg.ErrorMessage,
71
-
Region: strconv.Itoa(region.ID),
72
-
MonitorID: int64(monitors.ID),
65
+
ID: req.Msg.Id,
66
+
WorkspaceID: int64(ic.Monitor.WorkspaceID),
67
+
Timestamp: req.Msg.Timestamp,
68
+
Error: uint8(req.Msg.Error),
69
+
Region: strconv.Itoa(ic.Region.ID),
70
+
MonitorID: int64(ic.Monitor.ID),
73
71
Timing: req.Msg.Timing,
74
72
Latency: req.Msg.Latency,
75
73
CronTimestamp: req.Msg.CronTimestamp,
76
74
Trigger: "cron",
77
75
URI: req.Msg.Uri,
78
76
RequestStatus: req.Msg.RequestStatus,
79
-
80
-
Records: records,
81
-
}
82
-
if err := h.TbClient.SendEvent(ctx, data, dataSourceName); err != nil {
83
-
log.Ctx(ctx).Error().Err(err).Msg("failed to send event to tinybird")
84
-
}
85
-
_, err = h.db.NamedExec("UPDATE private_location SET last_seen_at = :last_seen_at WHERE id = :id", map[string]any{
86
-
"last_seen_at": time.Now().Unix(),
87
-
"id": region.ID,
88
-
})
89
-
if err != nil {
90
-
log.Ctx(ctx).Error().Err(err).Msg("failed to update private location")
77
+
Records: records,
91
78
}
79
+
80
+
h.sendEventAndUpdateLastSeen(ctx, data, tinybird.DatasourceDNS, ic.Region.ID)
81
+
92
82
return connect.NewResponse(&private_locationv1.IngestDNSResponse{}), nil
93
83
}
+211
apps/private-location/internal/server/ingest_dns_test.go
+211
apps/private-location/internal/server/ingest_dns_test.go
···
1
+
package server_test
2
+
3
+
import (
4
+
"context"
5
+
"net/http"
6
+
"testing"
7
+
8
+
"connectrpc.com/connect"
9
+
"github.com/openstatushq/openstatus/apps/private-location/internal/server"
10
+
"github.com/openstatushq/openstatus/apps/private-location/internal/tinybird"
11
+
private_locationv1 "github.com/openstatushq/openstatus/apps/private-location/proto/private_location/v1"
12
+
)
13
+
14
+
func TestIngestDNS_Unauthenticated(t *testing.T) {
15
+
h := server.NewPrivateLocationServer(testDB(), tinybird.NewClient(http.DefaultClient, ""))
16
+
17
+
req := connect.NewRequest(&private_locationv1.IngestDNSRequest{})
18
+
// No token header
19
+
resp, err := h.IngestDNS(context.Background(), req)
20
+
if err == nil {
21
+
t.Fatalf("expected error for missing token, got nil")
22
+
}
23
+
if connect.CodeOf(err) != connect.CodeUnauthenticated {
24
+
t.Errorf("expected unauthenticated code, got %v", connect.CodeOf(err))
25
+
}
26
+
if resp != nil {
27
+
t.Errorf("expected nil response, got %v", resp)
28
+
}
29
+
}
30
+
31
+
func TestIngestDNS_ValidationError_EmptyID(t *testing.T) {
32
+
h := server.NewPrivateLocationServer(testDB(), tinybird.NewClient(http.DefaultClient, ""))
33
+
34
+
req := connect.NewRequest(&private_locationv1.IngestDNSRequest{
35
+
Id: "",
36
+
Timestamp: 1234567890,
37
+
})
38
+
req.Header().Set("openstatus-token", "my-secret-key")
39
+
40
+
resp, err := h.IngestDNS(context.Background(), req)
41
+
if err == nil {
42
+
t.Fatalf("expected error for validation failure, got nil")
43
+
}
44
+
if connect.CodeOf(err) != connect.CodeInvalidArgument {
45
+
t.Errorf("expected invalid argument code, got %v", connect.CodeOf(err))
46
+
}
47
+
if resp != nil {
48
+
t.Errorf("expected nil response, got %v", resp)
49
+
}
50
+
}
51
+
52
+
func TestIngestDNS_ValidationError_InvalidTimestamp(t *testing.T) {
53
+
h := server.NewPrivateLocationServer(testDB(), tinybird.NewClient(http.DefaultClient, ""))
54
+
55
+
req := connect.NewRequest(&private_locationv1.IngestDNSRequest{
56
+
Id: "dns-123",
57
+
Timestamp: 0, // Invalid - must be positive
58
+
})
59
+
req.Header().Set("openstatus-token", "my-secret-key")
60
+
61
+
resp, err := h.IngestDNS(context.Background(), req)
62
+
if err == nil {
63
+
t.Fatalf("expected error for validation failure, got nil")
64
+
}
65
+
if connect.CodeOf(err) != connect.CodeInvalidArgument {
66
+
t.Errorf("expected invalid argument code, got %v", connect.CodeOf(err))
67
+
}
68
+
if resp != nil {
69
+
t.Errorf("expected nil response, got %v", resp)
70
+
}
71
+
}
72
+
73
+
func TestIngestDNS_ValidationError_NegativeLatency(t *testing.T) {
74
+
h := server.NewPrivateLocationServer(testDB(), tinybird.NewClient(http.DefaultClient, ""))
75
+
76
+
req := connect.NewRequest(&private_locationv1.IngestDNSRequest{
77
+
Id: "dns-123",
78
+
Latency: -100,
79
+
Timestamp: 1234567890,
80
+
})
81
+
req.Header().Set("openstatus-token", "my-secret-key")
82
+
83
+
resp, err := h.IngestDNS(context.Background(), req)
84
+
if err == nil {
85
+
t.Fatalf("expected error for validation failure, got nil")
86
+
}
87
+
if connect.CodeOf(err) != connect.CodeInvalidArgument {
88
+
t.Errorf("expected invalid argument code, got %v", connect.CodeOf(err))
89
+
}
90
+
if resp != nil {
91
+
t.Errorf("expected nil response, got %v", resp)
92
+
}
93
+
}
94
+
95
+
func TestIngestDNS_DBError(t *testing.T) {
96
+
h := server.NewPrivateLocationServer(testDB(), tinybird.NewClient(http.DefaultClient, ""))
97
+
98
+
req := connect.NewRequest(&private_locationv1.IngestDNSRequest{
99
+
Id: "nonexistent-monitor",
100
+
Timestamp: 1234567890,
101
+
})
102
+
req.Header().Set("openstatus-token", "invalid-token")
103
+
104
+
resp, err := h.IngestDNS(context.Background(), req)
105
+
if err == nil {
106
+
t.Fatalf("expected error for db failure, got nil")
107
+
}
108
+
if connect.CodeOf(err) != connect.CodeInternal {
109
+
t.Errorf("expected internal code, got %v", connect.CodeOf(err))
110
+
}
111
+
if resp != nil {
112
+
t.Errorf("expected nil response, got %v", resp)
113
+
}
114
+
}
115
+
116
+
func TestIngestDNS_MonitorNotExist(t *testing.T) {
117
+
h := server.NewPrivateLocationServer(testDB(), tinybird.NewClient(http.DefaultClient, ""))
118
+
119
+
req := connect.NewRequest(&private_locationv1.IngestDNSRequest{
120
+
Id: "nonexistent-monitor",
121
+
Timestamp: 1234567890,
122
+
})
123
+
req.Header().Set("openstatus-token", "my-secret-key")
124
+
125
+
resp, err := h.IngestDNS(context.Background(), req)
126
+
if err == nil {
127
+
t.Fatalf("expected error for monitor not found, got nil")
128
+
}
129
+
if connect.CodeOf(err) != connect.CodeInternal {
130
+
t.Errorf("expected internal code, got %v", connect.CodeOf(err))
131
+
}
132
+
if resp != nil {
133
+
t.Errorf("expected nil response, got %v", resp)
134
+
}
135
+
}
136
+
137
+
func TestIngestDNS_MonitorExist(t *testing.T) {
138
+
h := server.NewPrivateLocationServer(testDB(), getTBClient(context.Background()))
139
+
140
+
req := connect.NewRequest(&private_locationv1.IngestDNSRequest{
141
+
Id: "5",
142
+
Timestamp: 1234567890,
143
+
Latency: 50,
144
+
CronTimestamp: 1234567800,
145
+
Uri: "dns://example.com",
146
+
Timing: "50ms",
147
+
Records: nil,
148
+
})
149
+
req.Header().Set("openstatus-token", "my-secret-key")
150
+
151
+
resp, err := h.IngestDNS(context.Background(), req)
152
+
if err != nil {
153
+
t.Fatalf("expected nil error, got %v", err)
154
+
}
155
+
if resp == nil {
156
+
t.Errorf("expected not nil response, got nil")
157
+
}
158
+
}
159
+
160
+
func TestIngestDNS_WithRecords(t *testing.T) {
161
+
h := server.NewPrivateLocationServer(testDB(), getTBClient(context.Background()))
162
+
163
+
req := connect.NewRequest(&private_locationv1.IngestDNSRequest{
164
+
Id: "5",
165
+
Timestamp: 1234567890,
166
+
Latency: 50,
167
+
CronTimestamp: 1234567800,
168
+
Uri: "dns://example.com",
169
+
Timing: "50ms",
170
+
Records: map[string]*private_locationv1.Records{
171
+
"A": {
172
+
Record: []string{"192.168.1.1", "192.168.1.2"},
173
+
},
174
+
"AAAA": {
175
+
Record: []string{"::1"},
176
+
},
177
+
},
178
+
})
179
+
req.Header().Set("openstatus-token", "my-secret-key")
180
+
181
+
resp, err := h.IngestDNS(context.Background(), req)
182
+
if err != nil {
183
+
t.Fatalf("expected nil error, got %v", err)
184
+
}
185
+
if resp == nil {
186
+
t.Errorf("expected not nil response, got nil")
187
+
}
188
+
}
189
+
190
+
func TestIngestDNS_WithError(t *testing.T) {
191
+
h := server.NewPrivateLocationServer(testDB(), getTBClient(context.Background()))
192
+
193
+
req := connect.NewRequest(&private_locationv1.IngestDNSRequest{
194
+
Id: "5",
195
+
Timestamp: 1234567890,
196
+
Latency: 0,
197
+
CronTimestamp: 1234567800,
198
+
Uri: "dns://example.com",
199
+
Error: 1, // Error occurred
200
+
Message: "DNS resolution failed",
201
+
})
202
+
req.Header().Set("openstatus-token", "my-secret-key")
203
+
204
+
resp, err := h.IngestDNS(context.Background(), req)
205
+
if err != nil {
206
+
t.Fatalf("expected nil error, got %v", err)
207
+
}
208
+
if resp == nil {
209
+
t.Errorf("expected not nil response, got nil")
210
+
}
211
+
}
+20
-34
apps/private-location/internal/server/ingest_http.go
+20
-34
apps/private-location/internal/server/ingest_http.go
···
2
2
3
3
import (
4
4
"context"
5
-
"errors"
6
5
"strconv"
7
-
"time"
8
6
9
7
"connectrpc.com/connect"
10
-
"github.com/openstatushq/openstatus/apps/private-location/internal/database"
8
+
"github.com/openstatushq/openstatus/apps/private-location/internal/tinybird"
11
9
private_locationv1 "github.com/openstatushq/openstatus/apps/private-location/proto/private_location/v1"
12
-
"github.com/rs/zerolog/log"
13
10
)
14
11
15
12
type PingData struct {
···
34
31
}
35
32
36
33
func (h *privateLocationHandler) IngestHTTP(ctx context.Context, req *connect.Request[private_locationv1.IngestHTTPRequest]) (*connect.Response[private_locationv1.IngestHTTPResponse], error) {
34
+
event := ctx.Value("event")
35
+
if eventMap, ok := event.(map[string]any); ok && eventMap != nil {
36
+
eventMap["private_location"] = map[string]any{
37
+
"monitor_id": req.Msg.MonitorId,
38
+
}
39
+
ctx = context.WithValue(ctx, "event", eventMap)
40
+
}
41
+
37
42
token := req.Header().Get("openstatus-token")
38
43
if token == "" {
39
-
return nil, connect.NewError(connect.CodeUnauthenticated, errors.New("missing token"))
44
+
return nil, connect.NewError(connect.CodeUnauthenticated, ErrMissingToken)
40
45
}
41
46
42
-
dataSourceName := "ping_response__v8"
43
-
44
-
var monitors database.Monitor
45
-
err := h.db.Get(&monitors, "SELECT monitor.* FROM monitor JOIN private_location_to_monitor a ON monitor.id = a.monitor_id JOIN private_location b ON a.private_location_id = b.id WHERE b.token = ? AND monitor.deleted_at IS NULL and monitor.id = ?", token, req.Msg.MonitorId)
46
-
47
-
if err != nil {
48
-
log.Ctx(ctx).Error().Err(err).Msg("Failed to get monitors")
49
-
50
-
return nil, connect.NewError(connect.CodeInternal, err)
47
+
if err := ValidateIngestHTTPRequest(req.Msg); err != nil {
48
+
return nil, NewValidationError(err)
51
49
}
52
50
53
-
var region database.PrivateLocation
54
-
err = h.db.Get(®ion, "SELECT private_location.id FROM private_location join private_location_to_monitor a ON private_location.id = a.private_location_id WHERE a.monitor_id = ? AND private_location.token = ?", monitors.ID, token)
55
-
51
+
ic, err := h.getIngestContext(ctx, token, req.Msg.MonitorId)
56
52
if err != nil {
57
-
58
-
log.Ctx(ctx).Error().Err(err).Msg("Failed to get private location")
59
53
return nil, connect.NewError(connect.CodeInternal, err)
60
54
}
61
55
···
64
58
Latency: req.Msg.Latency,
65
59
StatusCode: int(req.Msg.StatusCode),
66
60
MonitorID: req.Msg.MonitorId,
67
-
Region: strconv.Itoa(region.ID),
68
-
WorkspaceID: strconv.Itoa(monitors.WorkspaceID),
61
+
Region: strconv.Itoa(ic.Region.ID),
62
+
WorkspaceID: strconv.Itoa(ic.Monitor.WorkspaceID),
69
63
Timestamp: req.Msg.Timestamp,
70
64
CronTimestamp: req.Msg.CronTimestamp,
71
-
URL: monitors.URL,
72
-
Method: monitors.Method,
65
+
URL: ic.Monitor.URL,
66
+
Method: ic.Monitor.Method,
73
67
Timing: req.Msg.Timing,
74
68
Headers: req.Msg.Headers,
75
69
Body: req.Msg.Body,
76
70
Trigger: "cron",
77
71
RequestStatus: req.Msg.RequestStatus,
78
-
Assertions: monitors.Assertions.String,
72
+
Assertions: ic.Monitor.Assertions.String,
79
73
}
80
-
if err := h.TbClient.SendEvent(ctx, data, dataSourceName); err != nil {
81
-
log.Ctx(ctx).Error().Err(err).Msg("failed to send event to tinybird")
82
-
}
83
-
_, err = h.db.NamedExec("UPDATE private_location SET last_seen_at = :last_seen_at WHERE id = :id", map[string]any{
84
-
"last_seen_at": time.Now().Unix(),
85
-
"id": region.ID,
86
-
})
87
-
if err != nil {
88
-
log.Ctx(ctx).Error().Err(err).Msg("failed to update private location")
89
-
}
74
+
75
+
h.sendEventAndUpdateLastSeen(ctx, data, tinybird.DatasourceHTTP, ic.Region.ID)
90
76
91
77
return connect.NewResponse(&private_locationv1.IngestHTTPResponse{}), nil
92
78
}
+120
-1
apps/private-location/internal/server/ingest_http_test.go
+120
-1
apps/private-location/internal/server/ingest_http_test.go
···
81
81
req := connect.NewRequest(&private_locationv1.IngestHTTPRequest{})
82
82
req.Header().Set("openstatus-token", "token123")
83
83
req.Msg.Id = "monitor1"
84
+
req.Msg.MonitorId = "nonexistent"
85
+
req.Msg.Timestamp = 1234567890
84
86
resp, err := h.IngestHTTP(context.Background(), req)
85
87
if err == nil {
86
88
t.Fatalf("expected error for db failure, got nil")
···
99
101
req := connect.NewRequest(&private_locationv1.IngestHTTPRequest{})
100
102
req.Header().Set("openstatus-token", "my-secret-key")
101
103
req.Msg.Id = "monitor1"
104
+
req.Msg.MonitorId = "nonexistent"
105
+
req.Msg.Timestamp = 1234567890
102
106
resp, err := h.IngestHTTP(context.Background(), req)
103
107
if err == nil {
104
108
t.Fatalf("expected error for db failure, got nil")
···
118
122
req.Header().Set("openstatus-token", "my-secret-key")
119
123
req.Msg.Id = "monitor1"
120
124
req.Msg.MonitorId = "5"
125
+
req.Msg.Timestamp = 1234567890
121
126
resp, err := h.IngestHTTP(context.Background(), req)
122
127
if err != nil {
123
128
t.Fatalf("expected nil error, got %v", err)
124
129
}
125
130
126
131
if resp == nil {
127
-
t.Errorf("expected not response, got %v", resp)
132
+
t.Errorf("expected not nil response, got %v", resp)
133
+
}
134
+
}
135
+
136
+
func TestIngestHTTP_ValidationError_EmptyMonitorID(t *testing.T) {
137
+
h := server.NewPrivateLocationServer(testDB(), getTBClient(context.Background()))
138
+
139
+
req := connect.NewRequest(&private_locationv1.IngestHTTPRequest{
140
+
MonitorId: "",
141
+
Timestamp: 1234567890,
142
+
})
143
+
req.Header().Set("openstatus-token", "my-secret-key")
144
+
145
+
resp, err := h.IngestHTTP(context.Background(), req)
146
+
if err == nil {
147
+
t.Fatalf("expected error for validation failure, got nil")
148
+
}
149
+
if connect.CodeOf(err) != connect.CodeInvalidArgument {
150
+
t.Errorf("expected invalid argument code, got %v", connect.CodeOf(err))
151
+
}
152
+
if resp != nil {
153
+
t.Errorf("expected nil response, got %v", resp)
154
+
}
155
+
}
156
+
157
+
func TestIngestHTTP_ValidationError_InvalidTimestamp(t *testing.T) {
158
+
h := server.NewPrivateLocationServer(testDB(), getTBClient(context.Background()))
159
+
160
+
req := connect.NewRequest(&private_locationv1.IngestHTTPRequest{
161
+
MonitorId: "5",
162
+
Timestamp: 0,
163
+
})
164
+
req.Header().Set("openstatus-token", "my-secret-key")
165
+
166
+
resp, err := h.IngestHTTP(context.Background(), req)
167
+
if err == nil {
168
+
t.Fatalf("expected error for validation failure, got nil")
169
+
}
170
+
if connect.CodeOf(err) != connect.CodeInvalidArgument {
171
+
t.Errorf("expected invalid argument code, got %v", connect.CodeOf(err))
172
+
}
173
+
if resp != nil {
174
+
t.Errorf("expected nil response, got %v", resp)
175
+
}
176
+
}
177
+
178
+
func TestIngestHTTP_ValidationError_NegativeLatency(t *testing.T) {
179
+
h := server.NewPrivateLocationServer(testDB(), getTBClient(context.Background()))
180
+
181
+
req := connect.NewRequest(&private_locationv1.IngestHTTPRequest{
182
+
MonitorId: "5",
183
+
Latency: -100,
184
+
Timestamp: 1234567890,
185
+
})
186
+
req.Header().Set("openstatus-token", "my-secret-key")
187
+
188
+
resp, err := h.IngestHTTP(context.Background(), req)
189
+
if err == nil {
190
+
t.Fatalf("expected error for validation failure, got nil")
191
+
}
192
+
if connect.CodeOf(err) != connect.CodeInvalidArgument {
193
+
t.Errorf("expected invalid argument code, got %v", connect.CodeOf(err))
194
+
}
195
+
if resp != nil {
196
+
t.Errorf("expected nil response, got %v", resp)
197
+
}
198
+
}
199
+
200
+
func TestIngestHTTP_WithFullData(t *testing.T) {
201
+
h := server.NewPrivateLocationServer(testDB(), getTBClient(context.Background()))
202
+
203
+
req := connect.NewRequest(&private_locationv1.IngestHTTPRequest{
204
+
Id: "request-1",
205
+
MonitorId: "5",
206
+
Timestamp: 1234567890,
207
+
Latency: 150,
208
+
CronTimestamp: 1234567800,
209
+
Url: "https://example.com/api",
210
+
StatusCode: 200,
211
+
Timing: "150ms",
212
+
Body: `{"status": "ok"}`,
213
+
Headers: `{"Content-Type": "application/json"}`,
214
+
})
215
+
req.Header().Set("openstatus-token", "my-secret-key")
216
+
217
+
resp, err := h.IngestHTTP(context.Background(), req)
218
+
if err != nil {
219
+
t.Fatalf("expected nil error, got %v", err)
220
+
}
221
+
if resp == nil {
222
+
t.Errorf("expected not nil response, got nil")
223
+
}
224
+
}
225
+
226
+
func TestIngestHTTP_WithError(t *testing.T) {
227
+
h := server.NewPrivateLocationServer(testDB(), getTBClient(context.Background()))
228
+
229
+
req := connect.NewRequest(&private_locationv1.IngestHTTPRequest{
230
+
Id: "request-1",
231
+
MonitorId: "5",
232
+
Timestamp: 1234567890,
233
+
Latency: 0,
234
+
CronTimestamp: 1234567800,
235
+
Url: "https://example.com/api",
236
+
Error: 1,
237
+
Message: "Connection timeout",
238
+
})
239
+
req.Header().Set("openstatus-token", "my-secret-key")
240
+
241
+
resp, err := h.IngestHTTP(context.Background(), req)
242
+
if err != nil {
243
+
t.Fatalf("expected nil error, got %v", err)
244
+
}
245
+
if resp == nil {
246
+
t.Errorf("expected not nil response, got nil")
128
247
}
129
248
}
+21
-31
apps/private-location/internal/server/ingest_tcp.go
+21
-31
apps/private-location/internal/server/ingest_tcp.go
···
2
2
3
3
import (
4
4
"context"
5
-
"errors"
6
5
"strconv"
7
-
"time"
8
6
9
7
"connectrpc.com/connect"
10
-
"github.com/openstatushq/openstatus/apps/private-location/internal/database"
8
+
"github.com/openstatushq/openstatus/apps/private-location/internal/tinybird"
11
9
private_locationv1 "github.com/openstatushq/openstatus/apps/private-location/proto/private_location/v1"
12
-
"github.com/rs/zerolog/log"
13
10
)
14
11
15
12
type TCPData struct {
···
34
31
func (h *privateLocationHandler) IngestTCP(ctx context.Context, req *connect.Request[private_locationv1.IngestTCPRequest]) (*connect.Response[private_locationv1.IngestTCPResponse], error) {
35
32
token := req.Header().Get("openstatus-token")
36
33
if token == "" {
37
-
return nil, connect.NewError(connect.CodeUnauthenticated, errors.New("missing token"))
34
+
return nil, connect.NewError(connect.CodeUnauthenticated, ErrMissingToken)
38
35
}
39
36
40
-
dataSourceName := "tcp_response__v0"
41
-
42
-
var monitors database.Monitor
43
-
err := h.db.Get(&monitors, "SELECT monitor.* FROM monitor JOIN private_location_to_monitor a ON monitor.id = a.monitor_id JOIN private_location b ON a.private_location_id = b.id WHERE b.token = ? AND monitor.deleted_at IS NULL and monitor.id = ?", token, req.Msg.Id)
37
+
if err := ValidateIngestTCPRequest(req.Msg); err != nil {
38
+
return nil, NewValidationError(err)
39
+
}
44
40
41
+
ic, err := h.getIngestContext(ctx, token, req.Msg.Id)
45
42
if err != nil {
46
43
return nil, connect.NewError(connect.CodeInternal, err)
47
44
}
48
45
49
-
var region database.PrivateLocation
50
-
err = h.db.Get(®ion, "SELECT private_location.id FROM private_location join private_location_to_monitor a ON private_location.id = a.private_location_id WHERE a.monitor_id = ? and private_location.token = ?", monitors.ID, token)
51
-
52
-
if err != nil {
53
-
return nil, connect.NewError(connect.CodeInternal, err)
46
+
event := ctx.Value("event")
47
+
if eventMap, ok := event.(map[string]any); ok && eventMap != nil {
48
+
eventMap["private_location"] = map[string]any{
49
+
"monitor_id": req.Msg.MonitorId,
50
+
}
51
+
ctx = context.WithValue(ctx, "event", eventMap)
54
52
}
55
53
56
54
data := TCPData{
57
-
ID: req.Msg.Id,
58
-
WorkspaceID: int64(monitors.WorkspaceID),
59
-
Timestamp: req.Msg.Timestamp,
60
-
Error: uint8(req.Msg.Error),
61
-
// ErrorMessage: req.Msg.ErrorMessage,
62
-
Region: strconv.Itoa(region.ID),
63
-
MonitorID: int64(monitors.ID),
55
+
ID: req.Msg.Id,
56
+
WorkspaceID: int64(ic.Monitor.WorkspaceID),
57
+
Timestamp: req.Msg.Timestamp,
58
+
Error: uint8(req.Msg.Error),
59
+
Region: strconv.Itoa(ic.Region.ID),
60
+
MonitorID: int64(ic.Monitor.ID),
64
61
Timing: req.Msg.Timing,
65
62
Latency: req.Msg.Latency,
66
63
CronTimestamp: req.Msg.CronTimestamp,
···
68
65
URI: req.Msg.Uri,
69
66
RequestStatus: req.Msg.RequestStatus,
70
67
}
71
-
if err := h.TbClient.SendEvent(ctx, data, dataSourceName); err != nil {
72
-
log.Ctx(ctx).Error().Err(err).Msg("failed to send event to tinybird")
73
-
}
74
-
_, err = h.db.NamedExec("UPDATE private_location SET last_seen_at = :last_seen_at WHERE id = :id", map[string]any{
75
-
"last_seen_at": time.Now().Unix(),
76
-
"id": region.ID,
77
-
})
78
-
if err != nil {
79
-
log.Ctx(ctx).Error().Err(err).Msg("failed to update private location")
80
-
}
68
+
69
+
h.sendEventAndUpdateLastSeen(ctx, data, tinybird.DatasourceTCP, ic.Region.ID)
70
+
81
71
return connect.NewResponse(&private_locationv1.IngestTCPResponse{}), nil
82
72
}
+131
apps/private-location/internal/server/ingest_tcp_test.go
+131
apps/private-location/internal/server/ingest_tcp_test.go
···
35
35
req := connect.NewRequest(&private_locationv1.IngestTCPRequest{})
36
36
req.Header().Set("openstatus-token", "token123")
37
37
req.Msg.Id = "monitor1"
38
+
req.Msg.Timestamp = 1234567890
38
39
resp, err := h.IngestTCP(context.Background(), req)
39
40
if err == nil {
40
41
t.Fatalf("expected error for db failure, got nil")
···
46
47
t.Errorf("expected nil response, got %v", resp)
47
48
}
48
49
}
50
+
51
+
func TestIngestTCP_ValidationError_EmptyID(t *testing.T) {
52
+
h := server.NewPrivateLocationServer(testDB(), tinybird.NewClient(http.DefaultClient, ""))
53
+
54
+
req := connect.NewRequest(&private_locationv1.IngestTCPRequest{
55
+
Id: "",
56
+
Timestamp: 1234567890,
57
+
})
58
+
req.Header().Set("openstatus-token", "my-secret-key")
59
+
60
+
resp, err := h.IngestTCP(context.Background(), req)
61
+
if err == nil {
62
+
t.Fatalf("expected error for validation failure, got nil")
63
+
}
64
+
if connect.CodeOf(err) != connect.CodeInvalidArgument {
65
+
t.Errorf("expected invalid argument code, got %v", connect.CodeOf(err))
66
+
}
67
+
if resp != nil {
68
+
t.Errorf("expected nil response, got %v", resp)
69
+
}
70
+
}
71
+
72
+
func TestIngestTCP_ValidationError_InvalidTimestamp(t *testing.T) {
73
+
h := server.NewPrivateLocationServer(testDB(), tinybird.NewClient(http.DefaultClient, ""))
74
+
75
+
req := connect.NewRequest(&private_locationv1.IngestTCPRequest{
76
+
Id: "tcp-123",
77
+
Timestamp: 0,
78
+
})
79
+
req.Header().Set("openstatus-token", "my-secret-key")
80
+
81
+
resp, err := h.IngestTCP(context.Background(), req)
82
+
if err == nil {
83
+
t.Fatalf("expected error for validation failure, got nil")
84
+
}
85
+
if connect.CodeOf(err) != connect.CodeInvalidArgument {
86
+
t.Errorf("expected invalid argument code, got %v", connect.CodeOf(err))
87
+
}
88
+
if resp != nil {
89
+
t.Errorf("expected nil response, got %v", resp)
90
+
}
91
+
}
92
+
93
+
func TestIngestTCP_ValidationError_NegativeLatency(t *testing.T) {
94
+
h := server.NewPrivateLocationServer(testDB(), tinybird.NewClient(http.DefaultClient, ""))
95
+
96
+
req := connect.NewRequest(&private_locationv1.IngestTCPRequest{
97
+
Id: "tcp-123",
98
+
Latency: -100,
99
+
Timestamp: 1234567890,
100
+
})
101
+
req.Header().Set("openstatus-token", "my-secret-key")
102
+
103
+
resp, err := h.IngestTCP(context.Background(), req)
104
+
if err == nil {
105
+
t.Fatalf("expected error for validation failure, got nil")
106
+
}
107
+
if connect.CodeOf(err) != connect.CodeInvalidArgument {
108
+
t.Errorf("expected invalid argument code, got %v", connect.CodeOf(err))
109
+
}
110
+
if resp != nil {
111
+
t.Errorf("expected nil response, got %v", resp)
112
+
}
113
+
}
114
+
115
+
func TestIngestTCP_MonitorNotExist(t *testing.T) {
116
+
h := server.NewPrivateLocationServer(testDB(), tinybird.NewClient(http.DefaultClient, ""))
117
+
118
+
req := connect.NewRequest(&private_locationv1.IngestTCPRequest{
119
+
Id: "nonexistent-monitor",
120
+
Timestamp: 1234567890,
121
+
})
122
+
req.Header().Set("openstatus-token", "my-secret-key")
123
+
124
+
resp, err := h.IngestTCP(context.Background(), req)
125
+
if err == nil {
126
+
t.Fatalf("expected error for monitor not found, got nil")
127
+
}
128
+
if connect.CodeOf(err) != connect.CodeInternal {
129
+
t.Errorf("expected internal code, got %v", connect.CodeOf(err))
130
+
}
131
+
if resp != nil {
132
+
t.Errorf("expected nil response, got %v", resp)
133
+
}
134
+
}
135
+
136
+
func TestIngestTCP_MonitorExist(t *testing.T) {
137
+
h := server.NewPrivateLocationServer(testDB(), getTBClient(context.Background()))
138
+
139
+
req := connect.NewRequest(&private_locationv1.IngestTCPRequest{
140
+
Id: "5",
141
+
Timestamp: 1234567890,
142
+
Latency: 50,
143
+
CronTimestamp: 1234567800,
144
+
Uri: "tcp://example.com:8080",
145
+
Timing: "50ms",
146
+
})
147
+
req.Header().Set("openstatus-token", "my-secret-key")
148
+
149
+
resp, err := h.IngestTCP(context.Background(), req)
150
+
if err != nil {
151
+
t.Fatalf("expected nil error, got %v", err)
152
+
}
153
+
if resp == nil {
154
+
t.Errorf("expected not nil response, got nil")
155
+
}
156
+
}
157
+
158
+
func TestIngestTCP_WithError(t *testing.T) {
159
+
h := server.NewPrivateLocationServer(testDB(), getTBClient(context.Background()))
160
+
161
+
req := connect.NewRequest(&private_locationv1.IngestTCPRequest{
162
+
Id: "5",
163
+
Timestamp: 1234567890,
164
+
Latency: 0,
165
+
CronTimestamp: 1234567800,
166
+
Uri: "tcp://example.com:8080",
167
+
Error: 1,
168
+
Message: "Connection refused",
169
+
})
170
+
req.Header().Set("openstatus-token", "my-secret-key")
171
+
172
+
resp, err := h.IngestTCP(context.Background(), req)
173
+
if err != nil {
174
+
t.Fatalf("expected nil error, got %v", err)
175
+
}
176
+
if resp == nil {
177
+
t.Errorf("expected not nil response, got nil")
178
+
}
179
+
}
+117
-30
apps/private-location/internal/server/monitors.go
+117
-30
apps/private-location/internal/server/monitors.go
···
4
4
"context"
5
5
"database/sql"
6
6
"encoding/json"
7
-
"errors"
8
7
"strconv"
9
8
10
9
"connectrpc.com/connect"
···
54
53
}
55
54
}
56
55
56
+
// Converts models.RecordComparator to proto RecordComparator
57
+
func convertRecordComparator(m models.RecordComparator) private_locationv1.RecordComparator {
58
+
switch m {
59
+
case models.RecordEquals:
60
+
return private_locationv1.RecordComparator_RECORD_COMPARATOR_EQUAL
61
+
case models.RecordNotEquals:
62
+
return private_locationv1.RecordComparator_RECORD_COMPARATOR_NOT_EQUAL
63
+
case models.RecordContains:
64
+
return private_locationv1.RecordComparator_RECORD_COMPARATOR_CONTAINS
65
+
case models.RecordNotContains:
66
+
return private_locationv1.RecordComparator_RECORD_COMPARATOR_NOT_CONTAINS
67
+
default:
68
+
return private_locationv1.RecordComparator_RECORD_COMPARATOR_UNSPECIFIED
69
+
}
70
+
}
71
+
57
72
// Helper to parse assertions
58
73
func ParseAssertions(assertions sql.NullString) (
59
74
statusAssertions []*private_locationv1.StatusCodeAssertion,
···
65
80
}
66
81
var rawAssertions []json.RawMessage
67
82
if err := json.Unmarshal([]byte(assertions.String), &rawAssertions); err != nil {
68
-
log.Printf("Failed to unmarshal assertions: %v", err)
83
+
log.Error().Err(err).Msg("failed to unmarshal assertions")
69
84
return
70
85
}
71
86
for _, a := range rawAssertions {
72
87
var assert models.Assertion
73
88
if err := json.Unmarshal(a, &assert); err != nil {
74
-
log.Printf("Failed to unmarshal assertion: %v", err)
89
+
log.Error().Err(err).Msg("failed to unmarshal assertion")
75
90
continue
76
91
}
77
92
switch assert.AssertionType {
78
93
case models.AssertionStatus:
79
94
var target models.StatusTarget
80
95
if err := json.Unmarshal(a, &target); err != nil {
81
-
log.Printf("Failed to unmarshal status target: %v", err)
96
+
log.Error().Err(err).Msg("failed to unmarshal status target")
82
97
continue
83
98
}
84
99
statusAssertions = append(statusAssertions, &private_locationv1.StatusCodeAssertion{
···
88
103
case models.AssertionHeader:
89
104
var target models.HeaderTarget
90
105
if err := json.Unmarshal(a, &target); err != nil {
91
-
log.Error().Err(err).Msg("unable to encode payload")
106
+
log.Error().Err(err).Msg("failed to unmarshal header target")
92
107
continue
93
108
}
94
109
headerAssertions = append(headerAssertions, &private_locationv1.HeaderAssertion{
···
99
114
case models.AssertionTextBody:
100
115
var target models.BodyString
101
116
if err := json.Unmarshal(a, &target); err != nil {
102
-
log.Printf("Failed to unmarshal body target: %v", err)
117
+
log.Error().Err(err).Msg("failed to unmarshal body target")
103
118
continue
104
119
}
105
120
bodyAssertions = append(bodyAssertions, &private_locationv1.BodyAssertion{
···
111
126
return
112
127
}
113
128
129
+
// Helper to parse DNS record assertions
130
+
func ParseRecordAssertions(assertions sql.NullString) []*private_locationv1.RecordAssertion {
131
+
if !assertions.Valid {
132
+
return nil
133
+
}
134
+
var rawAssertions []json.RawMessage
135
+
if err := json.Unmarshal([]byte(assertions.String), &rawAssertions); err != nil {
136
+
log.Error().Err(err).Msg("failed to unmarshal assertions")
137
+
return nil
138
+
}
139
+
var recordAssertions []*private_locationv1.RecordAssertion
140
+
for _, a := range rawAssertions {
141
+
var assert models.Assertion
142
+
if err := json.Unmarshal(a, &assert); err != nil {
143
+
log.Error().Err(err).Msg("failed to unmarshal assertion")
144
+
continue
145
+
}
146
+
if assert.AssertionType == models.AssertionDnsRecord {
147
+
var target models.RecordTarget
148
+
if err := json.Unmarshal(a, &target); err != nil {
149
+
log.Error().Err(err).Msg("failed to unmarshal record target")
150
+
continue
151
+
}
152
+
recordAssertions = append(recordAssertions, &private_locationv1.RecordAssertion{
153
+
Record: target.Key,
154
+
Comparator: convertRecordComparator(target.Comparator),
155
+
Target: target.Target,
156
+
})
157
+
}
158
+
}
159
+
return recordAssertions
160
+
}
161
+
114
162
func (h *privateLocationHandler) Monitors(ctx context.Context, req *connect.Request[private_locationv1.MonitorsRequest]) (*connect.Response[private_locationv1.MonitorsResponse], error) {
115
163
token := req.Header().Get("openstatus-token")
116
164
if token == "" {
117
-
return nil, connect.NewError(connect.CodeUnauthenticated, errors.New("missing token"))
165
+
return nil, connect.NewError(connect.CodeUnauthenticated, ErrMissingToken)
118
166
}
119
167
120
168
var monitors []database.Monitor
121
-
err := h.db.Select(&monitors, "SELECT monitor.* FROM monitor JOIN private_location_to_monitor a ON monitor.id = a.monitor_id JOIN private_location b ON a.private_location_id = b.id WHERE b.token = ? AND monitor.deleted_at IS NULL and monitor.active = 1", token)
169
+
err := h.db.Select(&monitors, "SELECT monitor.id, monitor.job_type, monitor.url, monitor.periodicity, monitor.method, monitor.body, monitor.timeout, monitor.degraded_after, monitor.follow_redirects, monitor.headers, monitor.assertions, monitor.workspace_id, monitor.retry FROM monitor JOIN private_location_to_monitor a ON monitor.id = a.monitor_id JOIN private_location b ON a.private_location_id = b.id WHERE b.token = ? AND monitor.deleted_at IS NULL and monitor.active = 1", token)
122
170
if err != nil {
123
171
return nil, connect.NewError(connect.CodeInternal, err)
124
172
}
125
-
173
+
var workspaceId int
126
174
var httpMonitors []*private_locationv1.HTTPMonitor
175
+
var tcpMonitors []*private_locationv1.TCPMonitor
176
+
var dnsMonitors []*private_locationv1.DNSMonitor
127
177
for _, monitor := range monitors {
128
-
if monitor.JobType != "http" {
129
-
continue
178
+
if workspaceId == 0 {
179
+
workspaceId = monitor.WorkspaceID
130
180
}
131
181
132
-
var headers []*private_locationv1.Headers
133
-
if err := json.Unmarshal([]byte(monitor.Headers), &headers); err != nil {
134
-
log.Ctx(ctx).Error().Err(err).Msg("unable to unmarshal headers")
135
-
headers = nil
182
+
switch monitor.JobType {
183
+
case database.JobTypeHTTP:
184
+
var headers []*private_locationv1.Headers
185
+
if err := json.Unmarshal([]byte(monitor.Headers), &headers); err != nil {
186
+
log.Ctx(ctx).Error().Err(err).Msg("unable to unmarshal headers")
187
+
headers = nil
188
+
}
189
+
190
+
statusAssertions, headerAssertions, bodyAssertions := ParseAssertions(monitor.Assertions)
191
+
192
+
httpMonitors = append(httpMonitors, &private_locationv1.HTTPMonitor{
193
+
Url: monitor.URL,
194
+
Periodicity: monitor.Periodicity,
195
+
Id: strconv.Itoa(monitor.ID),
196
+
Method: monitor.Method,
197
+
Body: monitor.Body,
198
+
Timeout: monitor.Timeout,
199
+
DegradedAt: &monitor.DegradedAfter.Int64,
200
+
Retry: int64(monitor.Retry),
201
+
FollowRedirects: monitor.FollowRedirects,
202
+
Headers: headers,
203
+
StatusCodeAssertions: statusAssertions,
204
+
HeaderAssertions: headerAssertions,
205
+
BodyAssertions: bodyAssertions,
206
+
})
207
+
208
+
case database.JobTypeTCP:
209
+
tcpMonitors = append(tcpMonitors, &private_locationv1.TCPMonitor{
210
+
Id: strconv.Itoa(monitor.ID),
211
+
Uri: monitor.URL,
212
+
Timeout: monitor.Timeout,
213
+
DegradedAt: &monitor.DegradedAfter.Int64,
214
+
Periodicity: monitor.Periodicity,
215
+
Retry: int64(monitor.Retry),
216
+
})
217
+
218
+
case database.JobTypeDNS:
219
+
recordAssertions := ParseRecordAssertions(monitor.Assertions)
220
+
dnsMonitors = append(dnsMonitors, &private_locationv1.DNSMonitor{
221
+
Id: strconv.Itoa(monitor.ID),
222
+
Uri: monitor.URL,
223
+
Timeout: monitor.Timeout,
224
+
DegradedAt: &monitor.DegradedAfter.Int64,
225
+
Periodicity: monitor.Periodicity,
226
+
Retry: int64(monitor.Retry),
227
+
RecordAssertions: recordAssertions,
228
+
})
136
229
}
230
+
}
137
231
138
-
statusAssertions, headerAssertions, bodyAssertions := ParseAssertions(monitor.Assertions)
139
232
140
-
httpMonitors = append(httpMonitors, &private_locationv1.HTTPMonitor{
141
-
Url: monitor.URL,
142
-
Periodicity: monitor.Periodicity,
143
-
Id: strconv.Itoa(monitor.ID),
144
-
Method: monitor.Method,
145
-
Body: monitor.Body,
146
-
Timeout: monitor.Timeout,
147
-
DegradedAt: &monitor.DegradedAfter.Int64,
148
-
FollowRedirects: monitor.FollowRedirects,
149
-
Headers: headers,
150
-
StatusCodeAssertions: statusAssertions,
151
-
HeaderAssertions: headerAssertions,
152
-
BodyAssertions: bodyAssertions,
153
-
})
233
+
event := ctx.Value("event")
234
+
if eventMap, ok := event.(map[string]any); ok && eventMap != nil {
235
+
eventMap["private_location"] = map[string]any{
236
+
"workspace_id": workspaceId,
237
+
}
238
+
ctx = context.WithValue(ctx, "event", eventMap)
154
239
}
155
240
156
241
return connect.NewResponse(&private_locationv1.MonitorsResponse{
157
242
HttpMonitors: httpMonitors,
243
+
TcpMonitors: tcpMonitors,
244
+
DnsMonitors: dnsMonitors,
158
245
}), nil
159
246
}
+532
apps/private-location/internal/server/monitors_test.go
+532
apps/private-location/internal/server/monitors_test.go
···
1
1
package server_test
2
2
3
3
import (
4
+
"context"
4
5
"database/sql"
5
6
"testing"
6
7
8
+
"connectrpc.com/connect"
7
9
"github.com/openstatushq/openstatus/apps/private-location/internal/server"
8
10
private_locationv1 "github.com/openstatushq/openstatus/apps/private-location/proto/private_location/v1"
9
11
)
···
55
57
t.Errorf("expected Comparator to be STRING_COMPARATOR_CONTAINS, got %v", got.Comparator)
56
58
}
57
59
}
60
+
61
+
func TestParseAssertions_InvalidJSON(t *testing.T) {
62
+
assertions := sql.NullString{
63
+
String: "not valid json",
64
+
Valid: true,
65
+
}
66
+
67
+
statusAssertions, headerAssertions, bodyAssertions := server.ParseAssertions(assertions)
68
+
69
+
if len(statusAssertions) != 0 || len(headerAssertions) != 0 || len(bodyAssertions) != 0 {
70
+
t.Errorf("expected empty assertions for invalid JSON, got status=%d, header=%d, body=%d",
71
+
len(statusAssertions), len(headerAssertions), len(bodyAssertions))
72
+
}
73
+
}
74
+
75
+
func TestParseAssertions_NullString(t *testing.T) {
76
+
assertions := sql.NullString{
77
+
String: "",
78
+
Valid: false,
79
+
}
80
+
81
+
statusAssertions, headerAssertions, bodyAssertions := server.ParseAssertions(assertions)
82
+
83
+
if len(statusAssertions) != 0 || len(headerAssertions) != 0 || len(bodyAssertions) != 0 {
84
+
t.Errorf("expected empty assertions for null string, got status=%d, header=%d, body=%d",
85
+
len(statusAssertions), len(headerAssertions), len(bodyAssertions))
86
+
}
87
+
}
88
+
89
+
func TestParseAssertions_HeaderAssertion(t *testing.T) {
90
+
input := `[{"version":"v1","type":"header","compare":"eq","key":"Content-Type","target":"application/json"}]`
91
+
assertions := sql.NullString{
92
+
String: input,
93
+
Valid: true,
94
+
}
95
+
96
+
_, headerAssertions, _ := server.ParseAssertions(assertions)
97
+
98
+
if len(headerAssertions) != 1 {
99
+
t.Fatalf("expected 1 header assertion, got %d", len(headerAssertions))
100
+
}
101
+
102
+
got := headerAssertions[0]
103
+
if got.Key != "Content-Type" {
104
+
t.Errorf("expected Key to be 'Content-Type', got '%s'", got.Key)
105
+
}
106
+
if got.Target != "application/json" {
107
+
t.Errorf("expected Target to be 'application/json', got '%s'", got.Target)
108
+
}
109
+
if got.Comparator != private_locationv1.StringComparator_STRING_COMPARATOR_EQUAL {
110
+
t.Errorf("expected Comparator to be STRING_COMPARATOR_EQUAL, got %v", got.Comparator)
111
+
}
112
+
}
113
+
114
+
func TestParseAssertions_MultipleAssertions(t *testing.T) {
115
+
input := `[
116
+
{"version":"v1","type":"status","compare":"eq","target":200},
117
+
{"version":"v1","type":"header","compare":"contains","key":"X-Request-Id","target":"req-"},
118
+
{"version":"v1","type":"textBody","compare":"notContains","target":"error"}
119
+
]`
120
+
assertions := sql.NullString{
121
+
String: input,
122
+
Valid: true,
123
+
}
124
+
125
+
statusAssertions, headerAssertions, bodyAssertions := server.ParseAssertions(assertions)
126
+
127
+
if len(statusAssertions) != 1 {
128
+
t.Errorf("expected 1 status assertion, got %d", len(statusAssertions))
129
+
}
130
+
if len(headerAssertions) != 1 {
131
+
t.Errorf("expected 1 header assertion, got %d", len(headerAssertions))
132
+
}
133
+
if len(bodyAssertions) != 1 {
134
+
t.Errorf("expected 1 body assertion, got %d", len(bodyAssertions))
135
+
}
136
+
}
137
+
138
+
func TestMonitors_Unauthenticated(t *testing.T) {
139
+
h := server.NewPrivateLocationServer(testDB(), getTBClient(context.Background()))
140
+
141
+
req := connect.NewRequest(&private_locationv1.MonitorsRequest{})
142
+
// No token header
143
+
resp, err := h.Monitors(context.Background(), req)
144
+
if err == nil {
145
+
t.Fatalf("expected error for missing token, got nil")
146
+
}
147
+
if connect.CodeOf(err) != connect.CodeUnauthenticated {
148
+
t.Errorf("expected unauthenticated code, got %v", connect.CodeOf(err))
149
+
}
150
+
if resp != nil {
151
+
t.Errorf("expected nil response, got %v", resp)
152
+
}
153
+
}
154
+
155
+
func TestMonitors_InvalidToken(t *testing.T) {
156
+
h := server.NewPrivateLocationServer(testDB(), getTBClient(context.Background()))
157
+
158
+
req := connect.NewRequest(&private_locationv1.MonitorsRequest{})
159
+
req.Header().Set("openstatus-token", "invalid-token")
160
+
161
+
resp, err := h.Monitors(context.Background(), req)
162
+
if err != nil {
163
+
t.Fatalf("expected no error for invalid token (just empty results), got %v", err)
164
+
}
165
+
if resp == nil {
166
+
t.Fatalf("expected non-nil response")
167
+
}
168
+
if len(resp.Msg.HttpMonitors) != 0 {
169
+
t.Errorf("expected 0 HTTP monitors for invalid token, got %d", len(resp.Msg.HttpMonitors))
170
+
}
171
+
if len(resp.Msg.TcpMonitors) != 0 {
172
+
t.Errorf("expected 0 TCP monitors for invalid token, got %d", len(resp.Msg.TcpMonitors))
173
+
}
174
+
if len(resp.Msg.DnsMonitors) != 0 {
175
+
t.Errorf("expected 0 DNS monitors for invalid token, got %d", len(resp.Msg.DnsMonitors))
176
+
}
177
+
}
178
+
179
+
func TestMonitors_ReturnsHTTPTCPAndDNSMonitors(t *testing.T) {
180
+
h := server.NewPrivateLocationServer(testDB(), getTBClient(context.Background()))
181
+
182
+
req := connect.NewRequest(&private_locationv1.MonitorsRequest{})
183
+
req.Header().Set("openstatus-token", "my-secret-key")
184
+
185
+
resp, err := h.Monitors(context.Background(), req)
186
+
if err != nil {
187
+
t.Fatalf("expected no error, got %v", err)
188
+
}
189
+
if resp == nil {
190
+
t.Fatalf("expected non-nil response")
191
+
}
192
+
193
+
// Should have HTTP monitor (monitor ID 5)
194
+
if len(resp.Msg.HttpMonitors) != 1 {
195
+
t.Errorf("expected 1 HTTP monitor, got %d", len(resp.Msg.HttpMonitors))
196
+
}
197
+
198
+
// Should have TCP monitor (monitor ID 6)
199
+
if len(resp.Msg.TcpMonitors) != 1 {
200
+
t.Errorf("expected 1 TCP monitor, got %d", len(resp.Msg.TcpMonitors))
201
+
}
202
+
203
+
// Should have DNS monitor (monitor ID 7)
204
+
if len(resp.Msg.DnsMonitors) != 1 {
205
+
t.Errorf("expected 1 DNS monitor, got %d", len(resp.Msg.DnsMonitors))
206
+
}
207
+
}
208
+
209
+
func TestMonitors_HTTPMonitorFields(t *testing.T) {
210
+
h := server.NewPrivateLocationServer(testDB(), getTBClient(context.Background()))
211
+
212
+
req := connect.NewRequest(&private_locationv1.MonitorsRequest{})
213
+
req.Header().Set("openstatus-token", "my-secret-key")
214
+
215
+
resp, err := h.Monitors(context.Background(), req)
216
+
if err != nil {
217
+
t.Fatalf("expected no error, got %v", err)
218
+
}
219
+
if len(resp.Msg.HttpMonitors) != 1 {
220
+
t.Fatalf("expected 1 HTTP monitor, got %d", len(resp.Msg.HttpMonitors))
221
+
}
222
+
223
+
httpMonitor := resp.Msg.HttpMonitors[0]
224
+
if httpMonitor.Id != "5" {
225
+
t.Errorf("expected ID '5', got '%s'", httpMonitor.Id)
226
+
}
227
+
if httpMonitor.Url != "https://openstat.us" {
228
+
t.Errorf("expected URL 'https://openstat.us', got '%s'", httpMonitor.Url)
229
+
}
230
+
if httpMonitor.Periodicity != "10m" {
231
+
t.Errorf("expected Periodicity '10m', got '%s'", httpMonitor.Periodicity)
232
+
}
233
+
if httpMonitor.Method != "GET" {
234
+
t.Errorf("expected Method 'GET', got '%s'", httpMonitor.Method)
235
+
}
236
+
if httpMonitor.Timeout != 45000 {
237
+
t.Errorf("expected Timeout 45000, got %d", httpMonitor.Timeout)
238
+
}
239
+
}
240
+
241
+
func TestMonitors_TCPMonitorFields(t *testing.T) {
242
+
h := server.NewPrivateLocationServer(testDB(), getTBClient(context.Background()))
243
+
244
+
req := connect.NewRequest(&private_locationv1.MonitorsRequest{})
245
+
req.Header().Set("openstatus-token", "my-secret-key")
246
+
247
+
resp, err := h.Monitors(context.Background(), req)
248
+
if err != nil {
249
+
t.Fatalf("expected no error, got %v", err)
250
+
}
251
+
if len(resp.Msg.TcpMonitors) != 1 {
252
+
t.Fatalf("expected 1 TCP monitor, got %d", len(resp.Msg.TcpMonitors))
253
+
}
254
+
255
+
tcpMonitor := resp.Msg.TcpMonitors[0]
256
+
if tcpMonitor.Id != "6" {
257
+
t.Errorf("expected ID '6', got '%s'", tcpMonitor.Id)
258
+
}
259
+
if tcpMonitor.Uri != "tcp://db.example.com:5432" {
260
+
t.Errorf("expected URI 'tcp://db.example.com:5432', got '%s'", tcpMonitor.Uri)
261
+
}
262
+
if tcpMonitor.Periodicity != "5m" {
263
+
t.Errorf("expected Periodicity '5m', got '%s'", tcpMonitor.Periodicity)
264
+
}
265
+
if tcpMonitor.Timeout != 30000 {
266
+
t.Errorf("expected Timeout 30000, got %d", tcpMonitor.Timeout)
267
+
}
268
+
if tcpMonitor.Retry != 2 {
269
+
t.Errorf("expected Retry 2, got %d", tcpMonitor.Retry)
270
+
}
271
+
if tcpMonitor.DegradedAt == nil || *tcpMonitor.DegradedAt != 5000 {
272
+
t.Errorf("expected DegradedAt 5000, got %v", tcpMonitor.DegradedAt)
273
+
}
274
+
}
275
+
276
+
func TestParseRecordAssertions_DnsRecordContains(t *testing.T) {
277
+
input := `[{"version":"v1","type":"dnsRecord","key":"A","compare":"contains","target":"76.76.21.21"}]`
278
+
assertions := sql.NullString{
279
+
String: input,
280
+
Valid: true,
281
+
}
282
+
283
+
recordAssertions := server.ParseRecordAssertions(assertions)
284
+
285
+
if len(recordAssertions) != 1 {
286
+
t.Fatalf("expected 1 record assertion, got %d", len(recordAssertions))
287
+
}
288
+
289
+
got := recordAssertions[0]
290
+
if got.Record != "A" {
291
+
t.Errorf("expected Record to be 'A', got '%s'", got.Record)
292
+
}
293
+
if got.Target != "76.76.21.21" {
294
+
t.Errorf("expected Target to be '76.76.21.21', got '%s'", got.Target)
295
+
}
296
+
if got.Comparator != private_locationv1.RecordComparator_RECORD_COMPARATOR_CONTAINS {
297
+
t.Errorf("expected Comparator to be RECORD_COMPARATOR_CONTAINS, got %v", got.Comparator)
298
+
}
299
+
}
300
+
301
+
func TestParseRecordAssertions_DnsRecordEquals(t *testing.T) {
302
+
input := `[{"version":"v1","type":"dnsRecord","key":"CNAME","compare":"eq","target":"openstatus.dev"}]`
303
+
assertions := sql.NullString{
304
+
String: input,
305
+
Valid: true,
306
+
}
307
+
308
+
recordAssertions := server.ParseRecordAssertions(assertions)
309
+
310
+
if len(recordAssertions) != 1 {
311
+
t.Fatalf("expected 1 record assertion, got %d", len(recordAssertions))
312
+
}
313
+
314
+
got := recordAssertions[0]
315
+
if got.Record != "CNAME" {
316
+
t.Errorf("expected Record to be 'CNAME', got '%s'", got.Record)
317
+
}
318
+
if got.Target != "openstatus.dev" {
319
+
t.Errorf("expected Target to be 'openstatus.dev', got '%s'", got.Target)
320
+
}
321
+
if got.Comparator != private_locationv1.RecordComparator_RECORD_COMPARATOR_EQUAL {
322
+
t.Errorf("expected Comparator to be RECORD_COMPARATOR_EQUAL, got %v", got.Comparator)
323
+
}
324
+
}
325
+
326
+
func TestParseRecordAssertions_MultipleRecordTypes(t *testing.T) {
327
+
input := `[
328
+
{"version":"v1","type":"dnsRecord","key":"A","compare":"eq","target":"192.168.1.1"},
329
+
{"version":"v1","type":"dnsRecord","key":"AAAA","compare":"not_eq","target":"::1"},
330
+
{"version":"v1","type":"dnsRecord","key":"MX","compare":"not_contains","target":"spam"}
331
+
]`
332
+
assertions := sql.NullString{
333
+
String: input,
334
+
Valid: true,
335
+
}
336
+
337
+
recordAssertions := server.ParseRecordAssertions(assertions)
338
+
339
+
if len(recordAssertions) != 3 {
340
+
t.Fatalf("expected 3 record assertions, got %d", len(recordAssertions))
341
+
}
342
+
343
+
// Check A record
344
+
if recordAssertions[0].Record != "A" {
345
+
t.Errorf("expected first Record to be 'A', got '%s'", recordAssertions[0].Record)
346
+
}
347
+
if recordAssertions[0].Comparator != private_locationv1.RecordComparator_RECORD_COMPARATOR_EQUAL {
348
+
t.Errorf("expected first Comparator to be RECORD_COMPARATOR_EQUAL, got %v", recordAssertions[0].Comparator)
349
+
}
350
+
351
+
// Check AAAA record
352
+
if recordAssertions[1].Record != "AAAA" {
353
+
t.Errorf("expected second Record to be 'AAAA', got '%s'", recordAssertions[1].Record)
354
+
}
355
+
if recordAssertions[1].Comparator != private_locationv1.RecordComparator_RECORD_COMPARATOR_NOT_EQUAL {
356
+
t.Errorf("expected second Comparator to be RECORD_COMPARATOR_NOT_EQUAL, got %v", recordAssertions[1].Comparator)
357
+
}
358
+
359
+
// Check MX record
360
+
if recordAssertions[2].Record != "MX" {
361
+
t.Errorf("expected third Record to be 'MX', got '%s'", recordAssertions[2].Record)
362
+
}
363
+
if recordAssertions[2].Comparator != private_locationv1.RecordComparator_RECORD_COMPARATOR_NOT_CONTAINS {
364
+
t.Errorf("expected third Comparator to be RECORD_COMPARATOR_NOT_CONTAINS, got %v", recordAssertions[2].Comparator)
365
+
}
366
+
}
367
+
368
+
func TestParseRecordAssertions_InvalidJSON(t *testing.T) {
369
+
assertions := sql.NullString{
370
+
String: "not valid json",
371
+
Valid: true,
372
+
}
373
+
374
+
recordAssertions := server.ParseRecordAssertions(assertions)
375
+
376
+
if len(recordAssertions) != 0 {
377
+
t.Errorf("expected empty assertions for invalid JSON, got %d", len(recordAssertions))
378
+
}
379
+
}
380
+
381
+
func TestParseRecordAssertions_NullString(t *testing.T) {
382
+
assertions := sql.NullString{
383
+
String: "",
384
+
Valid: false,
385
+
}
386
+
387
+
recordAssertions := server.ParseRecordAssertions(assertions)
388
+
389
+
if recordAssertions != nil {
390
+
t.Errorf("expected nil for null string, got %v", recordAssertions)
391
+
}
392
+
}
393
+
394
+
func TestParseRecordAssertions_MixedAssertionTypes(t *testing.T) {
395
+
// Test that only dnsRecord assertions are parsed, not other types
396
+
input := `[
397
+
{"version":"v1","type":"status","compare":"eq","target":200},
398
+
{"version":"v1","type":"dnsRecord","key":"A","compare":"contains","target":"192.168.1.1"},
399
+
{"version":"v1","type":"header","compare":"eq","key":"Content-Type","target":"application/json"}
400
+
]`
401
+
assertions := sql.NullString{
402
+
String: input,
403
+
Valid: true,
404
+
}
405
+
406
+
recordAssertions := server.ParseRecordAssertions(assertions)
407
+
408
+
if len(recordAssertions) != 1 {
409
+
t.Fatalf("expected 1 record assertion (only dnsRecord), got %d", len(recordAssertions))
410
+
}
411
+
412
+
if recordAssertions[0].Record != "A" {
413
+
t.Errorf("expected Record to be 'A', got '%s'", recordAssertions[0].Record)
414
+
}
415
+
}
416
+
417
+
func TestMonitors_DNSMonitorFields(t *testing.T) {
418
+
h := server.NewPrivateLocationServer(testDB(), getTBClient(context.Background()))
419
+
420
+
req := connect.NewRequest(&private_locationv1.MonitorsRequest{})
421
+
req.Header().Set("openstatus-token", "my-secret-key")
422
+
423
+
resp, err := h.Monitors(context.Background(), req)
424
+
if err != nil {
425
+
t.Fatalf("expected no error, got %v", err)
426
+
}
427
+
if len(resp.Msg.DnsMonitors) != 1 {
428
+
t.Fatalf("expected 1 DNS monitor, got %d", len(resp.Msg.DnsMonitors))
429
+
}
430
+
431
+
dnsMonitor := resp.Msg.DnsMonitors[0]
432
+
if dnsMonitor.Id != "7" {
433
+
t.Errorf("expected ID '7', got '%s'", dnsMonitor.Id)
434
+
}
435
+
if dnsMonitor.Uri != "openstatus.dev" {
436
+
t.Errorf("expected URI 'openstatus.dev', got '%s'", dnsMonitor.Uri)
437
+
}
438
+
if dnsMonitor.Periodicity != "5m" {
439
+
t.Errorf("expected Periodicity '5m', got '%s'", dnsMonitor.Periodicity)
440
+
}
441
+
if dnsMonitor.Timeout != 30000 {
442
+
t.Errorf("expected Timeout 30000, got %d", dnsMonitor.Timeout)
443
+
}
444
+
if dnsMonitor.Retry != 2 {
445
+
t.Errorf("expected Retry 2, got %d", dnsMonitor.Retry)
446
+
}
447
+
if dnsMonitor.DegradedAt == nil || *dnsMonitor.DegradedAt != 3000 {
448
+
t.Errorf("expected DegradedAt 3000, got %v", dnsMonitor.DegradedAt)
449
+
}
450
+
}
451
+
452
+
func TestParseRecordAssertions_EmptyArray(t *testing.T) {
453
+
assertions := sql.NullString{
454
+
String: "[]",
455
+
Valid: true,
456
+
}
457
+
458
+
recordAssertions := server.ParseRecordAssertions(assertions)
459
+
460
+
if len(recordAssertions) != 0 {
461
+
t.Errorf("expected empty slice for empty JSON array, got %d", len(recordAssertions))
462
+
}
463
+
}
464
+
465
+
func TestParseRecordAssertions_UnknownComparator(t *testing.T) {
466
+
input := `[{"version":"v1","type":"dnsRecord","key":"A","compare":"unknown_comparator","target":"192.168.1.1"}]`
467
+
assertions := sql.NullString{
468
+
String: input,
469
+
Valid: true,
470
+
}
471
+
472
+
recordAssertions := server.ParseRecordAssertions(assertions)
473
+
474
+
if len(recordAssertions) != 1 {
475
+
t.Fatalf("expected 1 record assertion, got %d", len(recordAssertions))
476
+
}
477
+
478
+
got := recordAssertions[0]
479
+
if got.Comparator != private_locationv1.RecordComparator_RECORD_COMPARATOR_UNSPECIFIED {
480
+
t.Errorf("expected Comparator to be RECORD_COMPARATOR_UNSPECIFIED for unknown comparator, got %v", got.Comparator)
481
+
}
482
+
}
483
+
484
+
func TestParseRecordAssertions_UnknownRecordType(t *testing.T) {
485
+
input := `[{"version":"v1","type":"dnsRecord","key":"INVALID_RECORD_TYPE","compare":"eq","target":"test"}]`
486
+
assertions := sql.NullString{
487
+
String: input,
488
+
Valid: true,
489
+
}
490
+
491
+
recordAssertions := server.ParseRecordAssertions(assertions)
492
+
493
+
if len(recordAssertions) != 1 {
494
+
t.Fatalf("expected 1 record assertion, got %d", len(recordAssertions))
495
+
}
496
+
497
+
got := recordAssertions[0]
498
+
// The record type is passed through as-is, even if invalid
499
+
if got.Record != "INVALID_RECORD_TYPE" {
500
+
t.Errorf("expected Record to be 'INVALID_RECORD_TYPE', got '%s'", got.Record)
501
+
}
502
+
}
503
+
504
+
func TestParseRecordAssertions_MissingRequiredFields(t *testing.T) {
505
+
// Missing "key" field
506
+
inputMissingKey := `[{"version":"v1","type":"dnsRecord","compare":"eq","target":"test"}]`
507
+
assertions := sql.NullString{
508
+
String: inputMissingKey,
509
+
Valid: true,
510
+
}
511
+
512
+
recordAssertions := server.ParseRecordAssertions(assertions)
513
+
514
+
if len(recordAssertions) != 1 {
515
+
t.Fatalf("expected 1 record assertion, got %d", len(recordAssertions))
516
+
}
517
+
518
+
// Missing key results in empty string
519
+
if recordAssertions[0].Record != "" {
520
+
t.Errorf("expected empty Record for missing key, got '%s'", recordAssertions[0].Record)
521
+
}
522
+
523
+
// Missing "compare" field
524
+
inputMissingCompare := `[{"version":"v1","type":"dnsRecord","key":"A","target":"test"}]`
525
+
assertions2 := sql.NullString{
526
+
String: inputMissingCompare,
527
+
Valid: true,
528
+
}
529
+
530
+
recordAssertions2 := server.ParseRecordAssertions(assertions2)
531
+
532
+
if len(recordAssertions2) != 1 {
533
+
t.Fatalf("expected 1 record assertion, got %d", len(recordAssertions2))
534
+
}
535
+
536
+
// Missing compare results in unspecified comparator
537
+
if recordAssertions2[0].Comparator != private_locationv1.RecordComparator_RECORD_COMPARATOR_UNSPECIFIED {
538
+
t.Errorf("expected RECORD_COMPARATOR_UNSPECIFIED for missing compare, got %v", recordAssertions2[0].Comparator)
539
+
}
540
+
541
+
// Missing "target" field
542
+
inputMissingTarget := `[{"version":"v1","type":"dnsRecord","key":"A","compare":"eq"}]`
543
+
assertions3 := sql.NullString{
544
+
String: inputMissingTarget,
545
+
Valid: true,
546
+
}
547
+
548
+
recordAssertions3 := server.ParseRecordAssertions(assertions3)
549
+
550
+
if len(recordAssertions3) != 1 {
551
+
t.Fatalf("expected 1 record assertion, got %d", len(recordAssertions3))
552
+
}
553
+
554
+
// Missing target results in empty string
555
+
if recordAssertions3[0].Target != "" {
556
+
t.Errorf("expected empty Target for missing target, got '%s'", recordAssertions3[0].Target)
557
+
}
558
+
}
559
+
560
+
func TestMonitors_DNSMonitorWithRecordAssertions(t *testing.T) {
561
+
h := server.NewPrivateLocationServer(testDB(), getTBClient(context.Background()))
562
+
563
+
req := connect.NewRequest(&private_locationv1.MonitorsRequest{})
564
+
req.Header().Set("openstatus-token", "my-secret-key")
565
+
566
+
resp, err := h.Monitors(context.Background(), req)
567
+
if err != nil {
568
+
t.Fatalf("expected no error, got %v", err)
569
+
}
570
+
if len(resp.Msg.DnsMonitors) != 1 {
571
+
t.Fatalf("expected 1 DNS monitor, got %d", len(resp.Msg.DnsMonitors))
572
+
}
573
+
574
+
dnsMonitor := resp.Msg.DnsMonitors[0]
575
+
if len(dnsMonitor.RecordAssertions) != 1 {
576
+
t.Fatalf("expected 1 record assertion, got %d", len(dnsMonitor.RecordAssertions))
577
+
}
578
+
579
+
assertion := dnsMonitor.RecordAssertions[0]
580
+
if assertion.Record != "A" {
581
+
t.Errorf("expected Record 'A', got '%s'", assertion.Record)
582
+
}
583
+
if assertion.Target != "76.76.21.21" {
584
+
t.Errorf("expected Target '76.76.21.21', got '%s'", assertion.Target)
585
+
}
586
+
if assertion.Comparator != private_locationv1.RecordComparator_RECORD_COMPARATOR_CONTAINS {
587
+
t.Errorf("expected Comparator RECORD_COMPARATOR_CONTAINS, got %v", assertion.Comparator)
588
+
}
589
+
}
+10
-2
apps/private-location/internal/server/routes.go
+10
-2
apps/private-location/internal/server/routes.go
···
161
161
162
162
// healthHandler responds with the health status of the server.
163
163
func (s *Server) healthHandler(w http.ResponseWriter, r *http.Request) {
164
+
status := "ok"
165
+
httpStatus := http.StatusOK
164
166
167
+
// Check database connection
168
+
if err := s.db.PingContext(r.Context()); err != nil {
169
+
status = "degraded"
170
+
httpStatus = http.StatusServiceUnavailable
171
+
}
172
+
173
+
render.Status(r, httpStatus)
165
174
render.JSON(w, r, map[string]any{
166
-
"status": "ok",
175
+
"status": status,
167
176
})
168
-
render.Status(r, http.StatusOK)
169
177
}
+10
-1
apps/private-location/internal/server/server.go
+10
-1
apps/private-location/internal/server/server.go
···
31
31
32
32
// NewServer returns an HTTP server and a cleanup function to shutdown the log provider.
33
33
func NewServer() (*http.Server, func(context.Context)) {
34
-
port, _ := strconv.Atoi(os.Getenv("PORT"))
34
+
portStr := os.Getenv("PORT")
35
+
if portStr == "" {
36
+
portStr = "8080"
37
+
}
38
+
port, err := strconv.Atoi(portStr)
39
+
if err != nil {
40
+
fmt.Fprintf(os.Stderr, "invalid PORT value %q: %v\n", portStr, err)
41
+
os.Exit(1)
42
+
}
35
43
36
44
logger, logProvider := setupLogger()
37
45
···
56
64
if logProvider != nil {
57
65
logProvider.Shutdown(ctx)
58
66
}
67
+
database.Close()
59
68
}
60
69
61
70
return server, cleanup
+64
apps/private-location/internal/server/validation.go
+64
apps/private-location/internal/server/validation.go
···
1
+
package server
2
+
3
+
import (
4
+
"errors"
5
+
"fmt"
6
+
7
+
"connectrpc.com/connect"
8
+
private_locationv1 "github.com/openstatushq/openstatus/apps/private-location/proto/private_location/v1"
9
+
)
10
+
11
+
// Validation errors
12
+
var (
13
+
ErrEmptyMonitorID = errors.New("monitor_id is required")
14
+
ErrEmptyID = errors.New("id is required")
15
+
ErrInvalidLatency = errors.New("latency must be non-negative")
16
+
ErrInvalidTimestamp = errors.New("timestamp must be positive")
17
+
)
18
+
19
+
// ValidateIngestHTTPRequest validates an HTTP ingest request
20
+
func ValidateIngestHTTPRequest(req *private_locationv1.IngestHTTPRequest) error {
21
+
if req.MonitorId == "" {
22
+
return ErrEmptyMonitorID
23
+
}
24
+
if req.Latency < 0 {
25
+
return ErrInvalidLatency
26
+
}
27
+
if req.Timestamp <= 0 {
28
+
return ErrInvalidTimestamp
29
+
}
30
+
return nil
31
+
}
32
+
33
+
// ValidateIngestTCPRequest validates a TCP ingest request
34
+
func ValidateIngestTCPRequest(req *private_locationv1.IngestTCPRequest) error {
35
+
if req.Id == "" {
36
+
return ErrEmptyID
37
+
}
38
+
if req.Latency < 0 {
39
+
return ErrInvalidLatency
40
+
}
41
+
if req.Timestamp <= 0 {
42
+
return ErrInvalidTimestamp
43
+
}
44
+
return nil
45
+
}
46
+
47
+
// ValidateIngestDNSRequest validates a DNS ingest request
48
+
func ValidateIngestDNSRequest(req *private_locationv1.IngestDNSRequest) error {
49
+
if req.Id == "" {
50
+
return ErrEmptyID
51
+
}
52
+
if req.Latency < 0 {
53
+
return ErrInvalidLatency
54
+
}
55
+
if req.Timestamp <= 0 {
56
+
return ErrInvalidTimestamp
57
+
}
58
+
return nil
59
+
}
60
+
61
+
// NewValidationError creates a Connect error for validation failures
62
+
func NewValidationError(err error) *connect.Error {
63
+
return connect.NewError(connect.CodeInvalidArgument, fmt.Errorf("validation error: %w", err))
64
+
}
+271
apps/private-location/internal/server/validation_test.go
+271
apps/private-location/internal/server/validation_test.go
···
1
+
package server_test
2
+
3
+
import (
4
+
"testing"
5
+
6
+
"connectrpc.com/connect"
7
+
"github.com/openstatushq/openstatus/apps/private-location/internal/server"
8
+
private_locationv1 "github.com/openstatushq/openstatus/apps/private-location/proto/private_location/v1"
9
+
)
10
+
11
+
func TestValidateIngestHTTPRequest(t *testing.T) {
12
+
tests := []struct {
13
+
name string
14
+
req *private_locationv1.IngestHTTPRequest
15
+
wantErr error
16
+
}{
17
+
{
18
+
name: "valid request",
19
+
req: &private_locationv1.IngestHTTPRequest{
20
+
MonitorId: "monitor-123",
21
+
Latency: 100,
22
+
Timestamp: 1234567890,
23
+
},
24
+
wantErr: nil,
25
+
},
26
+
{
27
+
name: "valid request with zero latency",
28
+
req: &private_locationv1.IngestHTTPRequest{
29
+
MonitorId: "monitor-123",
30
+
Latency: 0,
31
+
Timestamp: 1234567890,
32
+
},
33
+
wantErr: nil,
34
+
},
35
+
{
36
+
name: "empty monitor_id",
37
+
req: &private_locationv1.IngestHTTPRequest{
38
+
MonitorId: "",
39
+
Latency: 100,
40
+
Timestamp: 1234567890,
41
+
},
42
+
wantErr: server.ErrEmptyMonitorID,
43
+
},
44
+
{
45
+
name: "negative latency",
46
+
req: &private_locationv1.IngestHTTPRequest{
47
+
MonitorId: "monitor-123",
48
+
Latency: -1,
49
+
Timestamp: 1234567890,
50
+
},
51
+
wantErr: server.ErrInvalidLatency,
52
+
},
53
+
{
54
+
name: "zero timestamp",
55
+
req: &private_locationv1.IngestHTTPRequest{
56
+
MonitorId: "monitor-123",
57
+
Latency: 100,
58
+
Timestamp: 0,
59
+
},
60
+
wantErr: server.ErrInvalidTimestamp,
61
+
},
62
+
{
63
+
name: "negative timestamp",
64
+
req: &private_locationv1.IngestHTTPRequest{
65
+
MonitorId: "monitor-123",
66
+
Latency: 100,
67
+
Timestamp: -1,
68
+
},
69
+
wantErr: server.ErrInvalidTimestamp,
70
+
},
71
+
}
72
+
73
+
for _, tt := range tests {
74
+
t.Run(tt.name, func(t *testing.T) {
75
+
err := server.ValidateIngestHTTPRequest(tt.req)
76
+
if err != tt.wantErr {
77
+
t.Errorf("ValidateIngestHTTPRequest() error = %v, wantErr %v", err, tt.wantErr)
78
+
}
79
+
})
80
+
}
81
+
}
82
+
83
+
func TestValidateIngestTCPRequest(t *testing.T) {
84
+
tests := []struct {
85
+
name string
86
+
req *private_locationv1.IngestTCPRequest
87
+
wantErr error
88
+
}{
89
+
{
90
+
name: "valid request",
91
+
req: &private_locationv1.IngestTCPRequest{
92
+
Id: "tcp-123",
93
+
Latency: 100,
94
+
Timestamp: 1234567890,
95
+
},
96
+
wantErr: nil,
97
+
},
98
+
{
99
+
name: "valid request with zero latency",
100
+
req: &private_locationv1.IngestTCPRequest{
101
+
Id: "tcp-123",
102
+
Latency: 0,
103
+
Timestamp: 1234567890,
104
+
},
105
+
wantErr: nil,
106
+
},
107
+
{
108
+
name: "empty id",
109
+
req: &private_locationv1.IngestTCPRequest{
110
+
Id: "",
111
+
Latency: 100,
112
+
Timestamp: 1234567890,
113
+
},
114
+
wantErr: server.ErrEmptyID,
115
+
},
116
+
{
117
+
name: "negative latency",
118
+
req: &private_locationv1.IngestTCPRequest{
119
+
Id: "tcp-123",
120
+
Latency: -1,
121
+
Timestamp: 1234567890,
122
+
},
123
+
wantErr: server.ErrInvalidLatency,
124
+
},
125
+
{
126
+
name: "zero timestamp",
127
+
req: &private_locationv1.IngestTCPRequest{
128
+
Id: "tcp-123",
129
+
Latency: 100,
130
+
Timestamp: 0,
131
+
},
132
+
wantErr: server.ErrInvalidTimestamp,
133
+
},
134
+
{
135
+
name: "negative timestamp",
136
+
req: &private_locationv1.IngestTCPRequest{
137
+
Id: "tcp-123",
138
+
Latency: 100,
139
+
Timestamp: -1,
140
+
},
141
+
wantErr: server.ErrInvalidTimestamp,
142
+
},
143
+
}
144
+
145
+
for _, tt := range tests {
146
+
t.Run(tt.name, func(t *testing.T) {
147
+
err := server.ValidateIngestTCPRequest(tt.req)
148
+
if err != tt.wantErr {
149
+
t.Errorf("ValidateIngestTCPRequest() error = %v, wantErr %v", err, tt.wantErr)
150
+
}
151
+
})
152
+
}
153
+
}
154
+
155
+
func TestValidateIngestDNSRequest(t *testing.T) {
156
+
tests := []struct {
157
+
name string
158
+
req *private_locationv1.IngestDNSRequest
159
+
wantErr error
160
+
}{
161
+
{
162
+
name: "valid request",
163
+
req: &private_locationv1.IngestDNSRequest{
164
+
Id: "dns-123",
165
+
Latency: 100,
166
+
Timestamp: 1234567890,
167
+
},
168
+
wantErr: nil,
169
+
},
170
+
{
171
+
name: "valid request with zero latency",
172
+
req: &private_locationv1.IngestDNSRequest{
173
+
Id: "dns-123",
174
+
Latency: 0,
175
+
Timestamp: 1234567890,
176
+
},
177
+
wantErr: nil,
178
+
},
179
+
{
180
+
name: "empty id",
181
+
req: &private_locationv1.IngestDNSRequest{
182
+
Id: "",
183
+
Latency: 100,
184
+
Timestamp: 1234567890,
185
+
},
186
+
wantErr: server.ErrEmptyID,
187
+
},
188
+
{
189
+
name: "negative latency",
190
+
req: &private_locationv1.IngestDNSRequest{
191
+
Id: "dns-123",
192
+
Latency: -1,
193
+
Timestamp: 1234567890,
194
+
},
195
+
wantErr: server.ErrInvalidLatency,
196
+
},
197
+
{
198
+
name: "zero timestamp",
199
+
req: &private_locationv1.IngestDNSRequest{
200
+
Id: "dns-123",
201
+
Latency: 100,
202
+
Timestamp: 0,
203
+
},
204
+
wantErr: server.ErrInvalidTimestamp,
205
+
},
206
+
{
207
+
name: "negative timestamp",
208
+
req: &private_locationv1.IngestDNSRequest{
209
+
Id: "dns-123",
210
+
Latency: 100,
211
+
Timestamp: -1,
212
+
},
213
+
wantErr: server.ErrInvalidTimestamp,
214
+
},
215
+
}
216
+
217
+
for _, tt := range tests {
218
+
t.Run(tt.name, func(t *testing.T) {
219
+
err := server.ValidateIngestDNSRequest(tt.req)
220
+
if err != tt.wantErr {
221
+
t.Errorf("ValidateIngestDNSRequest() error = %v, wantErr %v", err, tt.wantErr)
222
+
}
223
+
})
224
+
}
225
+
}
226
+
227
+
func TestNewValidationError(t *testing.T) {
228
+
tests := []struct {
229
+
name string
230
+
err error
231
+
wantCode connect.Code
232
+
wantContains string
233
+
}{
234
+
{
235
+
name: "empty monitor id error",
236
+
err: server.ErrEmptyMonitorID,
237
+
wantCode: connect.CodeInvalidArgument,
238
+
wantContains: "monitor_id is required",
239
+
},
240
+
{
241
+
name: "empty id error",
242
+
err: server.ErrEmptyID,
243
+
wantCode: connect.CodeInvalidArgument,
244
+
wantContains: "id is required",
245
+
},
246
+
{
247
+
name: "invalid latency error",
248
+
err: server.ErrInvalidLatency,
249
+
wantCode: connect.CodeInvalidArgument,
250
+
wantContains: "latency must be non-negative",
251
+
},
252
+
{
253
+
name: "invalid timestamp error",
254
+
err: server.ErrInvalidTimestamp,
255
+
wantCode: connect.CodeInvalidArgument,
256
+
wantContains: "timestamp must be positive",
257
+
},
258
+
}
259
+
260
+
for _, tt := range tests {
261
+
t.Run(tt.name, func(t *testing.T) {
262
+
connErr := server.NewValidationError(tt.err)
263
+
if connErr.Code() != tt.wantCode {
264
+
t.Errorf("NewValidationError() code = %v, want %v", connErr.Code(), tt.wantCode)
265
+
}
266
+
if connErr.Message() == "" {
267
+
t.Error("NewValidationError() message should not be empty")
268
+
}
269
+
})
270
+
}
271
+
}
+7
apps/private-location/internal/tinybird/client.go
+7
apps/private-location/internal/tinybird/client.go
···
12
12
"github.com/rs/zerolog/log"
13
13
)
14
14
15
+
// Datasource names for Tinybird events
16
+
const (
17
+
DatasourceHTTP = "ping_response__v8"
18
+
DatasourceTCP = "tcp_response__v0"
19
+
DatasourceDNS = "tcp_dns__v0"
20
+
)
21
+
15
22
func getBaseURL() string {
16
23
// Use local Tinybird container if available (Docker/self-hosted)
17
24
// https://www.tinybird.co/docs/api-reference
+7
-7
apps/private-location/proto/private_location/v1/assertions.pb.go
+7
-7
apps/private-location/proto/private_location/v1/assertions.pb.go
···
1
1
// Code generated by protoc-gen-go. DO NOT EDIT.
2
2
// versions:
3
-
// protoc-gen-go v1.36.10
3
+
// protoc-gen-go v1.36.11
4
4
// protoc (unknown)
5
5
// source: private_location/v1/assertions.proto
6
6
···
378
378
state protoimpl.MessageState `protogen:"open.v1"`
379
379
Record string `protobuf:"bytes,1,opt,name=record,proto3" json:"record,omitempty"`
380
380
Comparator RecordComparator `protobuf:"varint,2,opt,name=comparator,proto3,enum=private_location.v1.RecordComparator" json:"comparator,omitempty"`
381
-
Targert string `protobuf:"bytes,3,opt,name=targert,proto3" json:"targert,omitempty"`
381
+
Target string `protobuf:"bytes,3,opt,name=target,proto3" json:"target,omitempty"`
382
382
unknownFields protoimpl.UnknownFields
383
383
sizeCache protoimpl.SizeCache
384
384
}
···
427
427
return RecordComparator_RECORD_COMPARATOR_UNSPECIFIED
428
428
}
429
429
430
-
func (x *RecordAssertion) GetTargert() string {
430
+
func (x *RecordAssertion) GetTarget() string {
431
431
if x != nil {
432
-
return x.Targert
432
+
return x.Target
433
433
}
434
434
return ""
435
435
}
···
454
454
"\n" +
455
455
"comparator\x18\x02 \x01(\x0e2%.private_location.v1.StringComparatorR\n" +
456
456
"comparator\x12\x10\n" +
457
-
"\x03key\x18\x03 \x01(\tR\x03key\"\x8a\x01\n" +
457
+
"\x03key\x18\x03 \x01(\tR\x03key\"\x88\x01\n" +
458
458
"\x0fRecordAssertion\x12\x16\n" +
459
459
"\x06record\x18\x01 \x01(\tR\x06record\x12E\n" +
460
460
"\n" +
461
461
"comparator\x18\x02 \x01(\x0e2%.private_location.v1.RecordComparatorR\n" +
462
-
"comparator\x12\x18\n" +
463
-
"\atargert\x18\x03 \x01(\tR\atargert*\x8f\x02\n" +
462
+
"comparator\x12\x16\n" +
463
+
"\x06target\x18\x03 \x01(\tR\x06target*\x8f\x02\n" +
464
464
"\x10NumberComparator\x12!\n" +
465
465
"\x1dNUMBER_COMPARATOR_UNSPECIFIED\x10\x00\x12\x1b\n" +
466
466
"\x17NUMBER_COMPARATOR_EQUAL\x10\x01\x12\x1f\n" +
+1
-1
apps/private-location/proto/private_location/v1/dns_monitor.pb.go
+1
-1
apps/private-location/proto/private_location/v1/dns_monitor.pb.go
+1
-1
apps/private-location/proto/private_location/v1/http_monitor.pb.go
+1
-1
apps/private-location/proto/private_location/v1/http_monitor.pb.go
+1
-1
apps/private-location/proto/private_location/v1/private_location.pb.go
+1
-1
apps/private-location/proto/private_location/v1/private_location.pb.go
+1
-1
apps/private-location/proto/private_location/v1/tcp_monitor.pb.go
+1
-1
apps/private-location/proto/private_location/v1/tcp_monitor.pb.go