-
Notifications
You must be signed in to change notification settings - Fork 307
/
admin.go
150 lines (134 loc) · 4.21 KB
/
admin.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
package stash
import (
"database/sql"
"encoding/json"
"fmt"
"github.com/rudderlabs/rudder-server/jobsdb"
"github.com/rudderlabs/rudder-server/utils/misc"
)
type StashRpcHandler struct {
ReadOnlyJobsDB jobsdb.ReadonlyJobsDB
}
var prefix = "proc_error_jobs_"
// GetDSStats
/*
ProcErrorsByDestinationCount
================================================================================
│───────│─────────────│──────────────────────────────────│
│ COUNT │ DESTINATION │ ERROR │
│───────│─────────────│──────────────────────────────────│
│ 3 │ GA │ "server side identify is not on" │
│───────│─────────────│──────────────────────────────────│
*/
func (s *StashRpcHandler) GetDSStats(dsName string, result *string) (err error) {
defer func() {
if r := recover(); r != nil {
pkgLogger.Error(r)
err = fmt.Errorf("internal Rudder server error: %v", r)
}
}()
jobTableName := prefix + dsName
dbHandle, err := sql.Open("postgres", misc.GetConnectionString())
// skipcq: SCC-SA5001
defer func() { _ = dbHandle.Close() }()
if err != nil {
return err
}
results, execErr := s.getErrorCountByDest(dbHandle, jobTableName)
if execErr != nil {
return execErr
}
response, marshalErr := json.MarshalIndent(results, "", " ")
if marshalErr != nil {
*result = ""
} else {
*result = string(response)
}
return marshalErr
}
func (s *StashRpcHandler) GetDSList(_ string, result *string) (err error) {
defer func() {
if r := recover(); r != nil {
pkgLogger.Error(r)
err = fmt.Errorf("internal Rudder server error: %v", r)
}
}()
response, err := s.ReadOnlyJobsDB.GetDSListString()
*result = response
return nil
}
func (s *StashRpcHandler) GetDSFailedJobs(arg string, result *string) (err error) {
defer func() {
if r := recover(); r != nil {
pkgLogger.Error(r)
err = fmt.Errorf("internal Rudder server error: %v", r)
}
}()
response, err := s.ReadOnlyJobsDB.GetLatestFailedJobs(arg, prefix)
*result = response
return err
}
func (s *StashRpcHandler) GetJobByID(arg string, result *string) (err error) {
defer func() {
if r := recover(); r != nil {
pkgLogger.Error(r)
err = fmt.Errorf("internal Rudder server error: %v", r)
}
}()
response, err := s.ReadOnlyJobsDB.GetJobByID(arg, prefix)
*result = response
return err
}
func (s *StashRpcHandler) GetJobIDStatus(arg string, result *string) (err error) {
defer func() {
if r := recover(); r != nil {
pkgLogger.Error(r)
err = fmt.Errorf("internal Rudder server error: %v", r)
}
}()
response, err := s.ReadOnlyJobsDB.GetJobIDStatus(arg, prefix)
*result = response
return err
}
type DestinationCountResult struct {
Count int
DestName string
Error string
}
func (*StashRpcHandler) getErrorCountByDest(dbHandle *sql.DB, jobTableName string) ([]DestinationCountResult, error) {
results := make([]DestinationCountResult, 0)
uniqueSourceValsStmt := fmt.Sprintf(`select count(*) as count, custom_val as dest, parameters -> 'error' as error from %s group by custom_val, parameters -> 'error'`, jobTableName)
var rows *sql.Rows
var err error
rows, err = dbHandle.Query(uniqueSourceValsStmt)
if err != nil {
return results, err
}
defer func() { _ = rows.Close() }()
singleResult := DestinationCountResult{}
for rows.Next() {
err = rows.Scan(&singleResult.Count, &singleResult.DestName, &singleResult.Error)
if err != nil {
return results, err
}
results = append(results, singleResult)
}
if err = rows.Err(); err != nil {
return results, err
}
if err = rows.Close(); err != nil {
return results, err
}
return results, nil
}
func (s *StashRpcHandler) GetDSJobCount(arg string, result *string) (err error) {
defer func() {
if r := recover(); r != nil {
pkgLogger.Error(r)
err = fmt.Errorf("internal Rudder server error: %v", r)
}
}()
response, err := s.ReadOnlyJobsDB.GetJobSummaryCount(arg, prefix)
*result = response
return nil
}