[Feature] Endpoints inspector

CHANGELOG.md

@@ -1,6 +1,7 @@
 # Change Log
 
 ## [master](https://github.com/arangodb/kube-arangodb/tree/master) (N/A)
+- (Feature) Add CoreV1 Endpoints Inspector
 
 ## [1.2.11](https://github.com/arangodb/kube-arangodb/tree/1.2.11) (2022-04-30)
 - (Bugfix) Orphan PVC are not removed

pkg/deployment/deployment.go

@@ -226,7 +226,8 @@ func newDeploymentThrottle() throttle.Components {
 		10*time.Second, // Secret
 		10*time.Second, // Service
 		30*time.Second, // SA
-		30*time.Second) // ServiceMonitor
+		30*time.Second, // ServiceMonitor
+		15*time.Second) // Endpoints
 }
 
 // New creates a new Deployment from the given API object.

pkg/deployment/resources/inspector/endpoints.go

@@ -0,0 +1,185 @@
+//
+// DISCLAIMER
+//
+// Copyright 2016-2022 ArangoDB GmbH, Cologne, Germany
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+// Copyright holder is ArangoDB GmbH, Cologne, Germany
+//
+
+package inspector
+
+import (
+	"context"
+	"time"
+
+	"github.com/arangodb/kube-arangodb/pkg/util/errors"
+	"github.com/arangodb/kube-arangodb/pkg/util/globals"
+	"github.com/arangodb/kube-arangodb/pkg/util/k8sutil/inspector/throttle"
+	core "k8s.io/api/core/v1"
+	meta "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+func init() {
+	requireRegisterInspectorLoader(endpointsInspectorLoaderObj)
+}
+
+var endpointsInspectorLoaderObj = endpointsInspectorLoader{}
+
+type endpointsInspectorLoader struct {
+}
+
+func (p endpointsInspectorLoader) Component() throttle.Component {
+	return throttle.Endpoints
+}
+
+func (p endpointsInspectorLoader) Load(ctx context.Context, i *inspectorState) {
+	var q endpointsInspector
+	p.loadV1(ctx, i, &q)
+	i.endpoints = &q
+	q.state = i
+	q.last = time.Now()
+}
+
+func (p endpointsInspectorLoader) loadV1(ctx context.Context, i *inspectorState, q *endpointsInspector) {
+	var z endpointsInspectorV1
+
+	z.endpointsInspector = q
+
+	z.endpoints, z.err = p.getV1Endpoints(ctx, i)
+
+	q.v1 = &z
+}
+
+func (p endpointsInspectorLoader) getV1Endpoints(ctx context.Context, i *inspectorState) (map[string]*core.Endpoints, error) {
+	objs, err := p.getV1EndpointsList(ctx, i)
+	if err != nil {
+		return nil, err
+	}
+
+	r := make(map[string]*core.Endpoints, len(objs))
+
+	for id := range objs {
+		r[objs[id].GetName()] = objs[id]
+	}
+
+	return r, nil
+}
+
+func (p endpointsInspectorLoader) getV1EndpointsList(ctx context.Context, i *inspectorState) ([]*core.Endpoints, error) {
+	ctxChild, cancel := globals.GetGlobalTimeouts().Kubernetes().WithTimeout(ctx)
+	defer cancel()
+	obj, err := i.client.Kubernetes().CoreV1().Endpoints(i.namespace).List(ctxChild, meta.ListOptions{
+		Limit: globals.GetGlobals().Kubernetes().RequestBatchSize().Get(),
+	})
+
+	if err != nil {
+		return nil, err
+	}
+
+	items := obj.Items
+	cont := obj.Continue
+	var s = int64(len(items))
+
+	if z := obj.RemainingItemCount; z != nil {
+		s += *z
+	}
+
+	ptrs := make([]*core.Endpoints, 0, s)
+
+	for {
+		for id := range items {
+			ptrs = append(ptrs, &items[id])
+		}
+
+		if cont == "" {
+			break
+		}
+
+		items, cont, err = p.getV1EndpointsListRequest(ctx, i, cont)
+
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	return ptrs, nil
+}
+
+func (p endpointsInspectorLoader) getV1EndpointsListRequest(ctx context.Context, i *inspectorState, cont string) ([]core.Endpoints, string, error) {
+	ctxChild, cancel := globals.GetGlobalTimeouts().Kubernetes().WithTimeout(ctx)
+	defer cancel()
+	obj, err := i.client.Kubernetes().CoreV1().Endpoints(i.namespace).List(ctxChild, meta.ListOptions{
+		Limit:    globals.GetGlobals().Kubernetes().RequestBatchSize().Get(),
+		Continue: cont,
+	})
+
+	if err != nil {
+		return nil, "", err
+	}
+
+	return obj.Items, obj.Continue, err
+}
+
+func (p endpointsInspectorLoader) Verify(i *inspectorState) error {
+	return nil
+}
+
+func (p endpointsInspectorLoader) Copy(from, to *inspectorState, override bool) {
+	if to.endpoints != nil {
+		if !override {
+			return
+		}
+	}
+
+	to.endpoints = from.endpoints
+	to.endpoints.state = to
+}
+
+func (p endpointsInspectorLoader) Name() string {
+	return "endpoints"
+}
+
+type endpointsInspector struct {
+	state *inspectorState
+
+	last time.Time
+
+	v1 *endpointsInspectorV1
+}
+
+func (p *endpointsInspector) LastRefresh() time.Time {
+	return p.last
+}
+
+func (p *endpointsInspector) Refresh(ctx context.Context) error {
+	p.Throttle(p.state.throttles).Invalidate()
+	return p.state.refresh(ctx, endpointsInspectorLoaderObj)
+}
+
+func (p endpointsInspector) Throttle(c throttle.Components) throttle.Throttle {
+	return c.Endpoints()
+}
+
+func (p *endpointsInspector) validate() error {
+	if p == nil {
+		return errors.Newf("EndpointsInspector is nil")
+	}
+
+	if p.state == nil {
+		return errors.Newf("Parent is nil")
+	}
+
+	return p.v1.validate()
+}

pkg/deployment/resources/inspector/endpoints_v1.go

@@ -0,0 +1,140 @@
+//
+// DISCLAIMER
+//
+// Copyright 2016-2022 ArangoDB GmbH, Cologne, Germany
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+// Copyright holder is ArangoDB GmbH, Cologne, Germany
+//
+
+package inspector
+
+import (
+	"context"
+
+	"github.com/arangodb/kube-arangodb/pkg/util/errors"
+	ins "github.com/arangodb/kube-arangodb/pkg/util/k8sutil/inspector/endpoints/v1"
+	core "k8s.io/api/core/v1"
+	apiErrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+)
+
+func (p *endpointsInspector) V1() (ins.Inspector, error) {
+	if p.v1.err != nil {
+		return nil, p.v1.err
+	}
+
+	return p.v1, nil
+}
+
+type endpointsInspectorV1 struct {
+	endpointsInspector *endpointsInspector
+
+	endpoints map[string]*core.Endpoints
+	err       error
+}
+
+func (p *endpointsInspectorV1) Filter(filters ...ins.Filter) []*core.Endpoints {
+	z := p.ListSimple()
+
+	r := make([]*core.Endpoints, 0, len(z))
+
+	for _, o := range z {
+		if !ins.FilterObject(o, filters...) {
+			continue
+		}
+
+		r = append(r, o)
+	}
+
+	return r
+}
+
+func (p *endpointsInspectorV1) validate() error {
+	if p == nil {
+		return errors.Newf("EndpointsV1Inspector is nil")
+	}
+
+	if p.endpointsInspector == nil {
+		return errors.Newf("Parent is nil")
+	}
+
+	if p.endpoints == nil && p.err == nil {
+		return errors.Newf("Endpoints or err should be not nil")
+	}
+
+	if p.endpoints != nil && p.err != nil {
+		return errors.Newf("Endpoints or err cannot be not nil together")
+	}
+
+	return nil
+}
+
+func (p *endpointsInspectorV1) ListSimple() []*core.Endpoints {
+	var r []*core.Endpoints
+	for _, endpoints := range p.endpoints {
+		r = append(r, endpoints)
+	}
+
+	return r
+}
+
+func (p *endpointsInspectorV1) GetSimple(name string) (*core.Endpoints, bool) {
+	endpoints, ok := p.endpoints[name]
+	if !ok {
+		return nil, false
+	}
+
+	return endpoints, true
+}
+
+func (p *endpointsInspectorV1) Iterate(action ins.Action, filters ...ins.Filter) error {
+	for _, endpoints := range p.endpoints {
+		if err := p.iterateEndpoints(endpoints, action, filters...); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (p *endpointsInspectorV1) iterateEndpoints(endpoints *core.Endpoints, action ins.Action, filters ...ins.Filter) error {
+	for _, f := range filters {
+		if f == nil {
+			continue
+		}
+
+		if !f(endpoints) {
+			return nil
+		}
+	}
+
+	return action(endpoints)
+}
+
+func (p *endpointsInspectorV1) Read() ins.ReadInterface {
+	return p
+}
+
+func (p *endpointsInspectorV1) Get(ctx context.Context, name string, opts metav1.GetOptions) (*core.Endpoints, error) {
+	if s, ok := p.GetSimple(name); !ok {
+		return nil, apiErrors.NewNotFound(schema.GroupResource{
+			Group:    core.GroupName,
+			Resource: "endpoints",
+		}, name)
+	} else {
+		return s, nil
+	}
+}

pkg/deployment/resources/inspector/inspector.go

@@ -33,6 +33,7 @@ import (
 	"github.com/arangodb/kube-arangodb/pkg/util/k8sutil/inspector/arangoclustersynchronization"
 	"github.com/arangodb/kube-arangodb/pkg/util/k8sutil/inspector/arangomember"
 	"github.com/arangodb/kube-arangodb/pkg/util/k8sutil/inspector/arangotask"
+	"github.com/arangodb/kube-arangodb/pkg/util/k8sutil/inspector/endpoints"
 	"github.com/arangodb/kube-arangodb/pkg/util/k8sutil/inspector/node"
 	"github.com/arangodb/kube-arangodb/pkg/util/k8sutil/inspector/persistentvolumeclaim"
 	"github.com/arangodb/kube-arangodb/pkg/util/k8sutil/inspector/pod"
@@ -134,6 +135,7 @@ type inspectorState struct {
 	arangoMembers                 *arangoMembersInspector
 	arangoTasks                   *arangoTasksInspector
 	arangoClusterSynchronizations *arangoClusterSynchronizationsInspector
+	endpoints                     *endpointsInspector
 
 	throttles throttle.Components
 
@@ -142,6 +144,10 @@ type inspectorState struct {
 	initialised bool
 }
 
+func (i *inspectorState) Endpoints() endpoints.Definition {
+	return i.endpoints
+}
+
 func (i *inspectorState) Initialised() bool {
 	if i == nil {
 		return false
@@ -345,6 +351,10 @@ func (i *inspectorState) validate() error {
 		return err
 	}
 
+	if err := i.endpoints.validate(); err != nil {
+		return err
+	}
+
 	return nil
 }
 
@@ -365,6 +375,7 @@ func (i *inspectorState) copyCore() *inspectorState {
 		arangoClusterSynchronizations: i.arangoClusterSynchronizations,
 		throttles:                     i.throttles.Copy(),
 		versionInfo:                   i.versionInfo,
+		endpoints:                     i.endpoints,
 		logger:                        i.logger,
 	}
 }