Skip to content

Commit 591a806

Browse files
jrauh01jrauh01
committedNov 28, 2023
Comment metrics
1 parent cff1a65 commit 591a806

File tree

1 file changed

+51
-38
lines changed

1 file changed

+51
-38
lines changed
 

‎pkg/sync/metrics.go

+51-38
Original file line numberDiff line numberDiff line change
@@ -17,12 +17,14 @@ import (
1717
metricsv "k8s.io/metrics/pkg/client/clientset/versioned"
1818
)
1919

20+
// MetricSync syncs container and pod metrics to the database
2021
type MetricSync struct {
2122
metricsClientset *metricsv.Clientset
2223
db *database.DB
2324
logger *logging.Logger
2425
}
2526

27+
// NewMetricSync creates new MetricSync initialized with metricsClientset, database and logger
2628
func NewMetricSync(metricsClientset *metricsv.Clientset, db *database.DB, logger *logging.Logger) *MetricSync {
2729
return &MetricSync{
2830
metricsClientset: metricsClientset,
@@ -31,6 +33,31 @@ func NewMetricSync(metricsClientset *metricsv.Clientset, db *database.DB, logger
3133
}
3234
}
3335

36+
// podMetricUpsertStmt returns database upsert statement to upsert pod metrics
37+
func (ms *MetricSync) podMetricUpsertStmt() string {
38+
return fmt.Sprintf(
39+
"INSERT INTO %s (%s) VALUES (%s) ON DUPLICATE KEY UPDATE %s",
40+
"pod_metric",
41+
"reference_id, timestamp, cpu, memory, storage",
42+
":reference_id, :timestamp, :cpu, :memory, :storage",
43+
"timestamp=VALUES(timestamp), cpu=VALUES(cpu), memory=VALUES(memory), storage=VALUES(storage)",
44+
)
45+
}
46+
47+
// containerMetricUpsertStmt returns database upsert statement to upsert container metrics
48+
func (ms *MetricSync) containerMetricUpsertStmt() string {
49+
return fmt.Sprintf(
50+
"INSERT INTO %s (%s) VALUES (%s) ON DUPLICATE KEY UPDATE %s",
51+
"container_metric",
52+
"container_reference_id, pod_reference_id, timestamp, cpu, memory, storage",
53+
":container_reference_id, :pod_reference_id, :timestamp, :cpu, :memory, :storage",
54+
"timestamp=VALUES(timestamp), cpu=VALUES(cpu), memory=VALUES(memory), storage=VALUES(storage)",
55+
)
56+
}
57+
58+
// Run starts syncing the metrics to the database. Therefore, it gets a list of all pods
59+
// and the belonging containers together with their metrics from the API every minute.
60+
// The pod metrics are the container metrics summed up by pod.
3461
func (ms *MetricSync) Run(ctx context.Context) error {
3562

3663
ms.logger.Info("Starting sync")
@@ -85,23 +112,23 @@ func (ms *MetricSync) Run(ctx context.Context) error {
85112
select {
86113
case <-ctx.Done():
87114
return ctx.Err()
88-
case <-time.After(time.Second * 5):
89-
//case <-time.After(time.Minute):
115+
case <-time.After(time.Minute):
90116
}
91117
}
92118
})
93119

94120
g.Go(func() error {
95-
return ms.db.UpsertStreamedWithStatement(ctx, upsertPodMetrics, ms.podMetricUpsertStmt(), 5)
121+
return ms.db.UpsertStreamed(ctx, upsertPodMetrics, database.WithStatement(ms.podMetricUpsertStmt(), 5))
96122
})
97123

98124
g.Go(func() error {
99-
return ms.db.UpsertStreamedWithStatement(ctx, upsertContainerMetrics, ms.containerMetricUpsertStmt(), 6)
125+
return ms.db.UpsertStreamed(ctx, upsertContainerMetrics, database.WithStatement(ms.containerMetricUpsertStmt(), 6))
100126
})
101127

102128
return g.Wait()
103129
}
104130

131+
// Clean deletes metrics from the database if the belonging pod is deleted
105132
func (ms *MetricSync) Clean(ctx context.Context, deleteChannel <-chan contracts.KDelete) error {
106133

107134
g, ctx := errgroup.WithContext(ctx)
@@ -130,42 +157,24 @@ func (ms *MetricSync) Clean(ctx context.Context, deleteChannel <-chan contracts.
130157
})
131158

132159
g.Go(func() error {
133-
return ms.db.DeleteStreamedByField(ctx, &schema.PodMetric{}, "reference_id", deletesPod)
160+
return ms.db.DeleteStreamed(ctx, &schema.PodMetric{}, deletesPod, database.ByColumn("reference_id"))
134161
})
135162

136163
g.Go(func() error {
137-
return ms.db.DeleteStreamedByField(ctx, &schema.ContainerMetric{}, "pod_reference_id", deletesContainer)
164+
return ms.db.DeleteStreamed(ctx, &schema.ContainerMetric{}, deletesContainer, database.ByColumn("pod_reference_id"))
138165
})
139166

140167
return g.Wait()
141168
}
142169

143-
func (ms *MetricSync) podMetricUpsertStmt() string {
144-
return fmt.Sprintf(
145-
"INSERT INTO %s (%s) VALUES (%s) ON DUPLICATE KEY UPDATE %s",
146-
"pod_metric",
147-
"reference_id, timestamp, cpu, memory, storage",
148-
":reference_id, :timestamp, :cpu, :memory, :storage",
149-
"timestamp=VALUES(timestamp), cpu=VALUES(cpu), memory=VALUES(memory), storage=VALUES(storage)",
150-
)
151-
}
152-
153-
func (ms *MetricSync) containerMetricUpsertStmt() string {
154-
return fmt.Sprintf(
155-
"INSERT INTO %s (%s) VALUES (%s) ON DUPLICATE KEY UPDATE %s",
156-
"container_metric",
157-
"container_reference_id, pod_reference_id, timestamp, cpu, memory, storage",
158-
":container_reference_id, :pod_reference_id, :timestamp, :cpu, :memory, :storage",
159-
"timestamp=VALUES(timestamp), cpu=VALUES(cpu), memory=VALUES(memory), storage=VALUES(storage)",
160-
)
161-
}
162-
170+
// NodeMetricSync syncs node metrics to the database
163171
type NodeMetricSync struct {
164172
metricsClientset *metricsv.Clientset
165173
db *database.DB
166174
logger *logging.Logger
167175
}
168176

177+
// NewNodeMetricSync creates new NodeMetricSync initialized with metricsClientset, database and logger
169178
func NewNodeMetricSync(metricClientset *metricsv.Clientset, db *database.DB, logger *logging.Logger) *NodeMetricSync {
170179
return &NodeMetricSync{
171180
metricsClientset: metricClientset,
@@ -174,6 +183,19 @@ func NewNodeMetricSync(metricClientset *metricsv.Clientset, db *database.DB, log
174183
}
175184
}
176185

186+
// nodeMetricUpsertStmt returns database upsert statement to upsert node metrics
187+
func (nms *NodeMetricSync) nodeMetricUpsertStmt() string {
188+
return fmt.Sprintf(
189+
"INSERT INTO %s (%s) VALUES (%s) ON DUPLICATE KEY UPDATE %s",
190+
"node_metric",
191+
"node_id, timestamp, cpu, memory, storage",
192+
":node_id, :timestamp, :cpu, :memory, :storage",
193+
"timestamp=VALUES(timestamp), cpu=VALUES(cpu), memory=VALUES(memory), storage=VALUES(storage)",
194+
)
195+
}
196+
197+
// Run starts syncing the metrics to the database. Therefore, it gets a list of all nodes
198+
// and the belonging metrics
177199
func (nms *NodeMetricSync) Run(ctx context.Context) error {
178200

179201
g, ctx := errgroup.WithContext(ctx)
@@ -207,12 +229,13 @@ func (nms *NodeMetricSync) Run(ctx context.Context) error {
207229
})
208230

209231
g.Go(func() error {
210-
return nms.db.UpsertStreamedWithStatement(ctx, upsertNodeMetrics, nms.nodeMetricUpsertStmt(), 5)
232+
return nms.db.UpsertStreamed(ctx, upsertNodeMetrics, database.WithStatement(nms.nodeMetricUpsertStmt(), 5))
211233
})
212234

213235
return g.Wait()
214236
}
215237

238+
// Clean deletes metrics from the database if the belonging node is deleted
216239
func (nms *NodeMetricSync) Clean(ctx context.Context, deleteChannel <-chan contracts.KDelete) error {
217240

218241
g, ctx := errgroup.WithContext(ctx)
@@ -238,18 +261,8 @@ func (nms *NodeMetricSync) Clean(ctx context.Context, deleteChannel <-chan contr
238261
})
239262

240263
g.Go(func() error {
241-
return nms.db.DeleteStreamedByField(ctx, &schema.NodeMetric{}, "node_id", deletes)
264+
return nms.db.DeleteStreamed(ctx, &schema.NodeMetric{}, deletes, database.ByColumn("node_id"))
242265
})
243266

244267
return g.Wait()
245268
}
246-
247-
func (nms *NodeMetricSync) nodeMetricUpsertStmt() string {
248-
return fmt.Sprintf(
249-
"INSERT INTO %s (%s) VALUES (%s) ON DUPLICATE KEY UPDATE %s",
250-
"node_metric",
251-
"node_id, timestamp, cpu, memory, storage",
252-
":node_id, :timestamp, :cpu, :memory, :storage",
253-
"timestamp=VALUES(timestamp), cpu=VALUES(cpu), memory=VALUES(memory), storage=VALUES(storage)",
254-
)
255-
}

0 commit comments

Comments
 (0)
Please sign in to comment.