Skip to content

Commit 8d8f427

Browse files
kerthcetkerthcet
authored
[Part-I] Metrics Aggregator implementation (#418)
* Sync the pod metrics Signed-off-by: kerthcet <kerthcet@gmail.com> * Add background goroutines Signed-off-by: kerthcet <kerthcet@gmail.com> * fix comment Signed-off-by: kerthcet <kerthcet@gmail.com> --------- Signed-off-by: kerthcet <kerthcet@gmail.com>
1 parent ef5178d commit 8d8f427

19 files changed

Lines changed: 604 additions & 48 deletions

File tree

‎chart/templates/backends/llamacpp.yaml‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -63,6 +63,7 @@ spec:
6363
- "16"
6464
- --draft-min
6565
- "5"
66+
- --metrics
6667
startupProbe:
6768
periodSeconds: 10
6869
failureThreshold: 30

‎components/metrics-aggregator/cmd/main.go‎

Lines changed: 9 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@ package main
1919
import (
2020
"flag"
2121
"os"
22+
"time"
2223

2324
// Import all Kubernetes client auth plugins (e.g. Azure, GCP, OIDC, etc.)
2425
// to ensure that exec-entrypoint and run can make use of them.
@@ -43,13 +44,19 @@ var (
4344
setupLog = ctrl.Log.WithName("setup")
4445
)
4546

47+
const (
48+
defaultSyncInterval = 50 * time.Millisecond
49+
)
50+
4651
func init() {
4752
utilruntime.Must(clientgoscheme.AddToScheme(scheme))
4853

4954
//+kubebuilder:scaffold:scheme
5055
}
5156

5257
func main() {
58+
ctx := ctrl.SetupSignalHandler()
59+
5360
var metricsAddr string
5461
var enableLeaderElection bool
5562
var probeAddr string
@@ -89,7 +96,7 @@ func main() {
8996
os.Exit(1)
9097
}
9198

92-
agg := aggregator.NewAggregator()
99+
agg := aggregator.NewAggregator(ctx, defaultSyncInterval)
93100

94101
if err := controller.NewPodReconciler(
95102
mgr.GetClient(),
@@ -113,7 +120,7 @@ func main() {
113120
}
114121

115122
setupLog.Info("starting manager")
116-
if err := mgr.Start(ctrl.SetupSignalHandler()); err != nil {
123+
if err := mgr.Start(ctx); err != nil {
117124
setupLog.Error(err, "problem running manager")
118125
os.Exit(1)
119126
}

‎components/metrics-aggregator/config/manager/kustomization.yaml‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -4,5 +4,5 @@ apiVersion: kustomize.config.k8s.io/v1beta1
44
kind: Kustomization
55
images:
66
- name: controller
7-
newName: inftyai/metrics-aggregator
8-
newTag: main
7+
newName: inftyai/test
8+
newTag: ma-051702

‎components/metrics-aggregator/go.mod‎

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,9 @@ module github.com/inftyai/metrics-aggregator
33
go 1.23.0
44

55
require (
6+
github.com/prometheus/client_model v0.4.0
7+
github.com/prometheus/common v0.44.0
8+
github.com/redis/go-redis/v9 v9.8.0
69
k8s.io/api v0.28.3
710
k8s.io/apimachinery v0.28.3
811
k8s.io/client-go v0.28.3
@@ -12,8 +15,9 @@ require (
1215

1316
require (
1417
github.com/beorn7/perks v1.0.1 // indirect
15-
github.com/cespare/xxhash/v2 v2.2.0 // indirect
18+
github.com/cespare/xxhash/v2 v2.3.0 // indirect
1619
github.com/davecgh/go-spew v1.1.1 // indirect
20+
github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
1721
github.com/emicklei/go-restful/v3 v3.11.0 // indirect
1822
github.com/evanphx/json-patch/v5 v5.6.0 // indirect
1923
github.com/fsnotify/fsnotify v1.6.0 // indirect
@@ -39,8 +43,6 @@ require (
3943
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
4044
github.com/pkg/errors v0.9.1 // indirect
4145
github.com/prometheus/client_golang v1.16.0 // indirect
42-
github.com/prometheus/client_model v0.4.0 // indirect
43-
github.com/prometheus/common v0.44.0 // indirect
4446
github.com/prometheus/procfs v0.10.1 // indirect
4547
github.com/spf13/pflag v1.0.5 // indirect
4648
go.uber.org/multierr v1.11.0 // indirect

‎components/metrics-aggregator/go.sum‎

Lines changed: 10 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -3,12 +3,18 @@ github.com/benbjohnson/clock v1.3.0 h1:ip6w0uFQkncKQ979AypyG0ER7mqUSBdKLOgAle/AT
33
github.com/benbjohnson/clock v1.3.0/go.mod h1:J11/hYXuz8f4ySSvYwY0FKfm+ezbsZBKZxNJlLklBHA=
44
github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
55
github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
6-
github.com/cespare/xxhash/v2 v2.2.0 h1:DC2CZ1Ep5Y4k3ZQ899DldepgrayRUGE6BBZ/cd9Cj44=
7-
github.com/cespare/xxhash/v2 v2.2.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
6+
github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs=
7+
github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c=
8+
github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA=
9+
github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0=
10+
github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
11+
github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
812
github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
913
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
1014
github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
1115
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
16+
github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78=
17+
github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc=
1218
github.com/emicklei/go-restful/v3 v3.11.0 h1:rAQeMHw1c7zTmncogyy8VvRZwtkmkZ4FxERmMY4rD+g=
1319
github.com/emicklei/go-restful/v3 v3.11.0/go.mod h1:6n3XBCmQQb25CM2LCACGz8ukIrRry+4bhvbpWn3mrbc=
1420
github.com/evanphx/json-patch v5.6.0+incompatible h1:jBYDEEiFBPxA0v50tFdvOzQQTCvpL6mnFh5mB2/l16U=
@@ -96,6 +102,8 @@ github.com/prometheus/common v0.44.0 h1:+5BrQJwiBB9xsMygAB3TNvpQKOwlkc25LbISbrdO
96102
github.com/prometheus/common v0.44.0/go.mod h1:ofAIvZbQ1e/nugmZGz4/qCb9Ap1VoSTIO7x0VV9VvuY=
97103
github.com/prometheus/procfs v0.10.1 h1:kYK1Va/YMlutzCGazswoHKo//tZVlFpKYh+PymziUAg=
98104
github.com/prometheus/procfs v0.10.1/go.mod h1:nwNm2aOCAYw8uTR/9bWRREkZFxAUcWzPHWJq+XBB/FM=
105+
github.com/redis/go-redis/v9 v9.8.0 h1:q3nRvjrlge/6UD7eTu/DSg2uYiU2mCL0G/uzBWqhicI=
106+
github.com/redis/go-redis/v9 v9.8.0/go.mod h1:huWgSWd8mW6+m0VPhJjSSQ+d6Nh1VICQ6Q5lHuCH/Iw=
99107
github.com/rogpeppe/go-internal v1.10.0 h1:TMyTOH3F/DB16zRVcYyreMH6GnZZrwQVAoYjRBZyWFQ=
100108
github.com/rogpeppe/go-internal v1.10.0/go.mod h1:UQnix2H7Ngw/k4C5ijL5+65zddjncjaFoBhdsK/akog=
101109
github.com/spf13/pflag v1.0.5 h1:iy+VFUOCP1a+8yFto/drg2CJ5u0yRoB7fZw3DKv/JXA=

‎components/metrics-aggregator/pkg/aggregator/aggregator.go‎

Lines changed: 131 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -17,10 +17,17 @@ limitations under the License.
1717
package aggregator
1818

1919
import (
20+
"context"
21+
"fmt"
2022
"sync"
2123
"sync/atomic"
24+
"time"
2225

2326
corev1 "k8s.io/api/core/v1"
27+
28+
"github.com/inftyai/metrics-aggregator/pkg/backend"
29+
"github.com/inftyai/metrics-aggregator/pkg/store"
30+
"github.com/inftyai/metrics-aggregator/pkg/util"
2431
)
2532

2633
func DefaultKeyFunc(pod *corev1.Pod) string {
@@ -34,31 +41,27 @@ type Aggregator struct {
3441
// the value is the PodWrapper.
3542
// TODO: Will all Pods share the same Map lead to performance issue once the number of pods is large.
3643
PodMap sync.Map
37-
// counter counts the number of pods in PodMap.
38-
counter atomic.Int32
3944
// KeyFunc represents the function to generate a key for the pod.
4045
KeyFunc func(pod *corev1.Pod) string
41-
}
42-
43-
type PodWrapper struct {
44-
// nsName is a string looks like "namespace/podName"
45-
nsName string
46-
// Pod is the pod object.
47-
// TODO: should we store the whole Pod object?
48-
pod *corev1.Pod
49-
}
5046

51-
func NewAggregator() *Aggregator {
52-
return &Aggregator{
53-
KeyFunc: DefaultKeyFunc,
54-
}
47+
// The system context controls when to stop all the goroutines.
48+
ctx context.Context
49+
// interval represents the how often to request the metrics from the pods.
50+
interval time.Duration
51+
// counter counts the number of pods in PodMap.
52+
counter atomic.Int32
53+
// store is the backend store to save the metrics.
54+
store store.Store
5555
}
5656

5757
func (a *Aggregator) AddPod(pod *corev1.Pod) {
5858
name := a.KeyFunc(pod)
59-
wrapper := &PodWrapper{nsName: name, pod: pod}
59+
wrapper := newPodWrapper(a.ctx, name, pod, a.store)
6060

6161
if _, ok := a.GetPod(name); !ok {
62+
wrapper.once.Do(func() {
63+
go wrapper.syncMetricsInLoop(a.interval)
64+
})
6265
a.PodMap.Store(name, wrapper)
6366
a.counter.Add(1)
6467
return
@@ -69,21 +72,128 @@ func (a *Aggregator) AddPod(pod *corev1.Pod) {
6972
a.PodMap.Store(name, wrapper)
7073
}
7174

72-
func (a *Aggregator) GetPod(nsName string) (*corev1.Pod, bool) {
73-
elem, ok := a.PodMap.Load(nsName)
75+
func (a *Aggregator) GetPod(name string) (*corev1.Pod, bool) {
76+
elem, ok := a.PodMap.Load(name)
7477
if !ok {
7578
return nil, false
7679
}
7780
return elem.(*PodWrapper).pod, true
7881
}
7982

80-
func (a *Aggregator) DeletePod(nsName string) {
81-
if _, ok := a.PodMap.Load(nsName); ok {
82-
a.PodMap.Delete(nsName)
83+
func (a *Aggregator) DeletePod(name string) {
84+
if wrapper, ok := a.PodMap.Load(name); ok {
85+
w := wrapper.(*PodWrapper)
86+
if err := w.stop(); err != nil {
87+
fmt.Printf("failed to stop the pod wrapper: %v", err)
88+
}
89+
a.PodMap.Delete(name)
8390
a.counter.Add(-1)
8491
}
8592
}
8693

8794
func (a *Aggregator) Len() int32 {
8895
return a.counter.Load()
8996
}
97+
98+
func NewAggregator(ctx context.Context, interval time.Duration) *Aggregator {
99+
// TODO: the store type should be optional.
100+
// FIXME: the redis host should be configurable.
101+
// store := store.NewRedisStore(ctx, "localhost")
102+
return &Aggregator{
103+
// We only have one aggregator, so it's ok to use the context directly.
104+
ctx: ctx,
105+
KeyFunc: DefaultKeyFunc,
106+
interval: interval,
107+
store: nil,
108+
}
109+
}
110+
111+
type PodWrapper struct {
112+
// ctx is the root context of the podWrapper.
113+
ctx context.Context
114+
// cancelFunc is the function to cancel the context of the background goroutine.
115+
cancelFunc context.CancelFunc
116+
// name is a string of the podWrapper.
117+
name string
118+
// the name of the model Pod is serving.
119+
modelName string
120+
// Pod is the pod object.
121+
// TODO: should we store the whole Pod object?
122+
pod *corev1.Pod
123+
// once ensures we only run the background goroutine for only once.
124+
once sync.Once
125+
// store used to save the metrics.
126+
store store.Store
127+
}
128+
129+
func (w *PodWrapper) syncMetricsInLoop(interval time.Duration) {
130+
ticket := time.NewTicker(interval)
131+
defer ticket.Stop()
132+
133+
for {
134+
select {
135+
case <-ticket.C:
136+
ep := metricEndpoint(w.pod)
137+
metrics, err := backend.QueryMetrics(ep)
138+
if err != nil {
139+
fmt.Printf("failed to query metrics from %s: %v, but continue.", ep, err)
140+
continue
141+
}
142+
// TODO: only for debug.
143+
for k, v := range metrics {
144+
fmt.Printf("%s metrics %s: %f\n", w.name, k, v)
145+
}
146+
if err := w.saveMetrics(metrics); err != nil {
147+
fmt.Printf("failed to save metrics to store: %v, but continue.", err)
148+
continue
149+
}
150+
case <-w.ctx.Done():
151+
fmt.Println("context done, stop the goroutine.")
152+
return
153+
}
154+
155+
// Avoid to count the request ping-pong time.
156+
ticket.Reset(interval)
157+
}
158+
}
159+
160+
func (w *PodWrapper) saveMetrics(metrics backend.MetricValues) error {
161+
runningQueueSize := metrics[backend.RunningQueueSize]
162+
waitingQueueSize := metrics[backend.WaitingQueueSize]
163+
// TODO: we may have more algorithms in the future based on different policies,
164+
// make it configurable.
165+
score := runningQueueSize*0.7 + waitingQueueSize*0.3
166+
// Only used in unit test to skip the store operation.
167+
if w.store != nil {
168+
return w.store.Insert(w.ctx, "LeastLatency:"+w.modelName, score, w.name)
169+
}
170+
return nil
171+
}
172+
173+
func (w *PodWrapper) stop() error {
174+
// Stop the background goroutine first.
175+
w.cancelFunc()
176+
// Only used in unit test to skip the store operation.
177+
if w.store != nil {
178+
// TODO: only we support more policies, we should refactor here.
179+
return w.store.Remove(w.ctx, "LeastLatency:"+w.modelName, w.name)
180+
}
181+
return nil
182+
}
183+
184+
func newPodWrapper(ctx context.Context, name string, pod *corev1.Pod, store store.Store) *PodWrapper {
185+
ctx, cancel := context.WithCancel(ctx)
186+
return &PodWrapper{
187+
ctx: ctx,
188+
cancelFunc: cancel,
189+
name: name,
190+
pod: pod,
191+
store: store,
192+
modelName: pod.Labels[util.ModelNameLabelKey],
193+
}
194+
}
195+
196+
// TODO: The http port is hardcoded now.
197+
func metricEndpoint(pod *corev1.Pod) string {
198+
return "http://" + pod.Status.PodIP + ":8080"
199+
}

‎components/metrics-aggregator/pkg/aggregator/aggregator_test.go‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,9 @@ limitations under the License.
1717
package aggregator
1818

1919
import (
20+
"context"
2021
"testing"
22+
"time"
2123

2224
corev1 "k8s.io/api/core/v1"
2325
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -37,7 +39,7 @@ func TestDefaultKeyFunc(t *testing.T) {
3739
}
3840

3941
func TestAggregator(t *testing.T) {
40-
agg := NewAggregator()
42+
agg := NewAggregator(context.Background(), 500*time.Millisecond)
4143
pod := &corev1.Pod{
4244
ObjectMeta: metav1.ObjectMeta{
4345
Name: "test-pod",

0 commit comments

Comments
 (0)