mirror of
https://github.com/VictoriaMetrics/VictoriaMetrics.git
synced 2024-11-21 14:44:00 +00:00
lib/promscrape/discovery/kubernetes: localize Bookmark parsing code
This is a follow-up for e772d1c920
This commit is contained in:
parent
ccfe944ce7
commit
41f641b132
9 changed files with 43 additions and 32 deletions
|
@ -31,7 +31,6 @@ type WatchEvent struct {
|
|||
type object interface {
|
||||
key() string
|
||||
getTargetLabels(aw *apiWatcher) []map[string]string
|
||||
resourceVersion() string
|
||||
}
|
||||
|
||||
// parseObjectFunc must parse object from the given data.
|
||||
|
@ -397,13 +396,13 @@ func (uw *urlWatcher) readObjectUpdateStream(r io.Reader) error {
|
|||
if err := d.Decode(&we); err != nil {
|
||||
return err
|
||||
}
|
||||
switch we.Type {
|
||||
case "ADDED", "MODIFIED":
|
||||
o, err := uw.parseObject(we.Object)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
key := o.key()
|
||||
switch we.Type {
|
||||
case "ADDED", "MODIFIED":
|
||||
uw.objectsByKey.update(key, o)
|
||||
labels := o.getTargetLabels(aw)
|
||||
swos := getScrapeWorkObjectsForLabels(aw.swcFunc, labels)
|
||||
|
@ -415,19 +414,44 @@ func (uw *urlWatcher) readObjectUpdateStream(r io.Reader) error {
|
|||
}
|
||||
uw.mu.Unlock()
|
||||
case "DELETED":
|
||||
o, err := uw.parseObject(we.Object)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
key := o.key()
|
||||
uw.objectsByKey.remove(key)
|
||||
uw.mu.Lock()
|
||||
delete(uw.swosByKey, key)
|
||||
uw.mu.Unlock()
|
||||
case "BOOKMARK":
|
||||
// See https://kubernetes.io/docs/reference/using-api/api-concepts/#watch-bookmarks
|
||||
uw.setResourceVersion(o.resourceVersion())
|
||||
bm, err := parseBookmark(we.Object)
|
||||
if err != nil {
|
||||
return fmt.Errorf("cannot parse bookmark from %q: %w", we.Object, err)
|
||||
}
|
||||
uw.setResourceVersion(bm.Metadata.ResourceVersion)
|
||||
default:
|
||||
return fmt.Errorf("unexpected WatchEvent type %q for role %q", we.Type, uw.role)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Bookmark is a bookmark from Kubernetes Watch API.
|
||||
// See https://kubernetes.io/docs/reference/using-api/api-concepts/#watch-bookmarks
|
||||
type Bookmark struct {
|
||||
Metadata struct {
|
||||
ResourceVersion string
|
||||
}
|
||||
}
|
||||
|
||||
func parseBookmark(data []byte) (*Bookmark, error) {
|
||||
var bm Bookmark
|
||||
if err := json.Unmarshal(data, &bm); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &bm, nil
|
||||
}
|
||||
|
||||
func getAPIPaths(role string, namespaces []string, selectors []Selector) []string {
|
||||
objectName := getObjectNameByRole(role)
|
||||
if objectName == "nodes" || len(namespaces) == 0 {
|
||||
|
|
|
@ -160,3 +160,15 @@ func TestGetAPIPaths(t *testing.T) {
|
|||
"/apis/networking.k8s.io/v1beta1/namespaces/y/ingresses?labelSelector=cde%2Cbaaa&fieldSelector=abc",
|
||||
})
|
||||
}
|
||||
|
||||
func TestParseBookmark(t *testing.T) {
|
||||
data := `{"kind": "Pod", "apiVersion": "v1", "metadata": {"resourceVersion": "12746"} }`
|
||||
bm, err := parseBookmark([]byte(data))
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected error: %s", err)
|
||||
}
|
||||
expectedResourceVersion := "12746"
|
||||
if bm.Metadata.ResourceVersion != expectedResourceVersion {
|
||||
t.Fatalf("unexpected resourceVersion; got %q; want %q", bm.Metadata.ResourceVersion, expectedResourceVersion)
|
||||
}
|
||||
}
|
||||
|
|
|
@ -8,7 +8,6 @@ import (
|
|||
//
|
||||
// See https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.17/#objectmeta-v1-meta
|
||||
type ObjectMeta struct {
|
||||
ResourceVersion string
|
||||
Name string
|
||||
Namespace string
|
||||
UID string
|
||||
|
|
|
@ -47,10 +47,6 @@ type Endpoints struct {
|
|||
Subsets []EndpointSubset
|
||||
}
|
||||
|
||||
func (eps *Endpoints) resourceVersion() string {
|
||||
return eps.Metadata.ResourceVersion
|
||||
}
|
||||
|
||||
// EndpointSubset implements k8s endpoint subset.
|
||||
//
|
||||
// See https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.17/#endpointsubset-v1-core
|
||||
|
|
|
@ -161,10 +161,6 @@ type EndpointSlice struct {
|
|||
Ports []EndpointPort
|
||||
}
|
||||
|
||||
func (eps *EndpointSlice) resourceVersion() string {
|
||||
return eps.Metadata.ResourceVersion
|
||||
}
|
||||
|
||||
// Endpoint implements kubernetes object endpoint for endpoint slice.
|
||||
// https://v1-17.docs.kubernetes.io/docs/reference/generated/kubernetes-api/v1.17/#endpoint-v1beta1-discovery-k8s-io
|
||||
type Endpoint struct {
|
||||
|
|
|
@ -45,10 +45,6 @@ type Ingress struct {
|
|||
Spec IngressSpec
|
||||
}
|
||||
|
||||
func (ig *Ingress) resourceVersion() string {
|
||||
return ig.Metadata.ResourceVersion
|
||||
}
|
||||
|
||||
// IngressSpec represents ingress spec in k8s.
|
||||
//
|
||||
// See https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.17/#ingressspec-v1beta1-extensions
|
||||
|
|
|
@ -48,10 +48,6 @@ type Node struct {
|
|||
Status NodeStatus
|
||||
}
|
||||
|
||||
func (n *Node) resourceVersion() string {
|
||||
return n.Metadata.ResourceVersion
|
||||
}
|
||||
|
||||
// NodeStatus represents NodeStatus from k8s API.
|
||||
//
|
||||
// See https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.17/#nodestatus-v1-core
|
||||
|
|
|
@ -50,10 +50,6 @@ type Pod struct {
|
|||
Status PodStatus
|
||||
}
|
||||
|
||||
func (p *Pod) resourceVersion() string {
|
||||
return p.Metadata.ResourceVersion
|
||||
}
|
||||
|
||||
// PodSpec implements k8s pod spec.
|
||||
//
|
||||
// See https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.17/#podspec-v1-core
|
||||
|
|
|
@ -47,10 +47,6 @@ type Service struct {
|
|||
Spec ServiceSpec
|
||||
}
|
||||
|
||||
func (s *Service) resourceVersion() string {
|
||||
return s.Metadata.ResourceVersion
|
||||
}
|
||||
|
||||
// ServiceSpec is k8s service spec.
|
||||
//
|
||||
// See https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.17/#servicespec-v1-core
|
||||
|
|
Loading…
Reference in a new issue