mirror of
https://github.com/weaveworks/scope.git
synced 2026-03-03 02:00:43 +00:00
Merge pull request #3014 from weaveworks/stop-fetching-replicasets
Stop fetching ReplicaSets and ReplicationControllers
This commit is contained in:
Roberto Bruggemann
GitHub
@@ -30,11 +30,9 @@ type Client interface {
|
||||
WalkPods(f func(Pod) error) error
|
||||
WalkServices(f func(Service) error) error
|
||||
WalkDeployments(f func(Deployment) error) error
|
||||
WalkReplicaSets(f func(ReplicaSet) error) error
|
||||
WalkDaemonSets(f func(DaemonSet) error) error
|
||||
WalkStatefulSets(f func(StatefulSet) error) error
|
||||
WalkCronJobs(f func(CronJob) error) error
|
||||
WalkReplicationControllers(f func(ReplicationController) error) error
|
||||
WalkNamespaces(f func(NamespaceResource) error) error
|
||||
|
||||
WatchPods(f func(Event, Pod))
|
||||
@@ -46,20 +44,18 @@ type Client interface {
|
||||
}
|
||||
|
||||
type client struct {
|
||||
quit chan struct{}
|
||||
resyncPeriod time.Duration
|
||||
client *kubernetes.Clientset
|
||||
podStore cache.Store
|
||||
serviceStore cache.Store
|
||||
deploymentStore cache.Store
|
||||
replicaSetStore cache.Store
|
||||
daemonSetStore cache.Store
|
||||
statefulSetStore cache.Store
|
||||
jobStore cache.Store
|
||||
cronJobStore cache.Store
|
||||
replicationControllerStore cache.Store
|
||||
nodeStore cache.Store
|
||||
namespaceStore cache.Store
|
||||
quit chan struct{}
|
||||
resyncPeriod time.Duration
|
||||
client *kubernetes.Clientset
|
||||
podStore cache.Store
|
||||
serviceStore cache.Store
|
||||
deploymentStore cache.Store
|
||||
daemonSetStore cache.Store
|
||||
statefulSetStore cache.Store
|
||||
jobStore cache.Store
|
||||
cronJobStore cache.Store
|
||||
nodeStore cache.Store
|
||||
namespaceStore cache.Store
|
||||
|
||||
podWatchesMutex sync.Mutex
|
||||
podWatches []func(Event, Pod)
|
||||
@@ -157,18 +153,16 @@ func NewClient(config ClientConfig) (Client, error) {
|
||||
result.podStore = result.setupStore(c.CoreV1Client.RESTClient(), "pods", &apiv1.Pod{}, podStore)
|
||||
|
||||
result.serviceStore = result.setupStore(c.CoreV1Client.RESTClient(), "services", &apiv1.Service{}, nil)
|
||||
result.replicationControllerStore = result.setupStore(c.CoreV1Client.RESTClient(), "replicationcontrollers", &apiv1.ReplicationController{}, nil)
|
||||
result.nodeStore = result.setupStore(c.CoreV1Client.RESTClient(), "nodes", &apiv1.Node{}, nil)
|
||||
result.namespaceStore = result.setupStore(c.CoreV1Client.RESTClient(), "namespaces", &apiv1.Namespace{}, nil)
|
||||
|
||||
// We list deployments here to check if this version of kubernetes is >= 1.2.
|
||||
// We would use NegotiateVersion, but Kubernetes 1.1 "supports"
|
||||
// extensions/v1beta1, but not deployments, replicasets or daemonsets.
|
||||
// extensions/v1beta1, but not deployments or daemonsets.
|
||||
if _, err := c.Extensions().Deployments(metav1.NamespaceAll).List(metav1.ListOptions{}); err != nil {
|
||||
log.Infof("Deployments, ReplicaSets and DaemonSets are not supported by this Kubernetes version: %v", err)
|
||||
log.Infof("Deployments and DaemonSets are not supported by this Kubernetes version: %v", err)
|
||||
} else {
|
||||
result.deploymentStore = result.setupStore(c.ExtensionsV1beta1Client.RESTClient(), "deployments", &apiextensionsv1beta1.Deployment{}, nil)
|
||||
result.replicaSetStore = result.setupStore(c.ExtensionsV1beta1Client.RESTClient(), "replicasets", &apiextensionsv1beta1.ReplicaSet{}, nil)
|
||||
result.daemonSetStore = result.setupStore(c.ExtensionsV1beta1Client.RESTClient(), "daemonsets", &apiextensionsv1beta1.DaemonSet{}, nil)
|
||||
}
|
||||
// CronJobs and StatefulSets were introduced later. Easiest to use the same technique.
|
||||
@@ -244,32 +238,6 @@ func (c *client) WalkDeployments(f func(Deployment) error) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// WalkReplicaSets calls f for each replica set
|
||||
func (c *client) WalkReplicaSets(f func(ReplicaSet) error) error {
|
||||
if c.replicaSetStore == nil {
|
||||
return nil
|
||||
}
|
||||
for _, m := range c.replicaSetStore.List() {
|
||||
rs := m.(*apiextensionsv1beta1.ReplicaSet)
|
||||
if err := f(NewReplicaSet(rs)); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
|
||||
}
|
||||
|
||||
// WalkReplicationcontrollers calls f for each replication controller
|
||||
func (c *client) WalkReplicationControllers(f func(ReplicationController) error) error {
|
||||
for _, m := range c.replicationControllerStore.List() {
|
||||
rc := m.(*apiv1.ReplicationController)
|
||||
if err := f(NewReplicationController(rc)); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// WalkDaemonSets calls f for each daemonset
|
||||
func (c *client) WalkDaemonSets(f func(DaemonSet) error) error {
|
||||
if c.daemonSetStore == nil {
|
||||
|
||||
@@ -74,81 +74,43 @@ func (r *Reporter) CapturePod(f func(xfer.Request, string, string, []string) xfe
|
||||
}
|
||||
}
|
||||
|
||||
// CaptureResource is exported for testing
|
||||
func (r *Reporter) CaptureResource(f func(xfer.Request, string, string, string) xfer.Response) func(xfer.Request) xfer.Response {
|
||||
// CaptureDeployment is exported for testing
|
||||
func (r *Reporter) CaptureDeployment(f func(xfer.Request, string, string) xfer.Response) func(xfer.Request) xfer.Response {
|
||||
return func(req xfer.Request) xfer.Response {
|
||||
var resource, uid string
|
||||
for _, parser := range []struct {
|
||||
res string
|
||||
f func(string) (string, bool)
|
||||
}{
|
||||
{report.Deployment, report.ParseDeploymentNodeID},
|
||||
{report.ReplicaSet, report.ParseReplicaSetNodeID},
|
||||
} {
|
||||
if u, ok := parser.f(req.NodeID); ok {
|
||||
resource, uid = parser.res, u
|
||||
break
|
||||
}
|
||||
}
|
||||
if resource == "" {
|
||||
uid, ok := report.ParseDeploymentNodeID(req.NodeID)
|
||||
if !ok {
|
||||
return xfer.ResponseErrorf("Invalid ID: %s", req.NodeID)
|
||||
}
|
||||
|
||||
switch resource {
|
||||
case report.Deployment:
|
||||
var deployment Deployment
|
||||
r.client.WalkDeployments(func(d Deployment) error {
|
||||
if d.UID() == uid {
|
||||
deployment = d
|
||||
}
|
||||
return nil
|
||||
})
|
||||
if deployment != nil {
|
||||
return f(req, "deployment", deployment.Namespace(), deployment.Name())
|
||||
}
|
||||
case report.ReplicaSet:
|
||||
var replicaSet ReplicaSet
|
||||
var res string
|
||||
r.client.WalkReplicaSets(func(r ReplicaSet) error {
|
||||
if r.UID() == uid {
|
||||
replicaSet = r
|
||||
res = "replicaset"
|
||||
}
|
||||
return nil
|
||||
})
|
||||
if replicaSet == nil {
|
||||
r.client.WalkReplicationControllers(func(r ReplicationController) error {
|
||||
if r.UID() == uid {
|
||||
replicaSet = ReplicaSet(r)
|
||||
res = "replicationcontroller"
|
||||
}
|
||||
return nil
|
||||
})
|
||||
}
|
||||
if replicaSet != nil {
|
||||
return f(req, res, replicaSet.Namespace(), replicaSet.Name())
|
||||
var deployment Deployment
|
||||
r.client.WalkDeployments(func(d Deployment) error {
|
||||
if d.UID() == uid {
|
||||
deployment = d
|
||||
}
|
||||
return nil
|
||||
})
|
||||
if deployment == nil {
|
||||
return xfer.ResponseErrorf("Deployment not found: %s", uid)
|
||||
}
|
||||
return xfer.ResponseErrorf("%s not found: %s", resource, uid)
|
||||
return f(req, deployment.Namespace(), deployment.Name())
|
||||
}
|
||||
}
|
||||
|
||||
// ScaleUp is the control to scale up a deployment
|
||||
func (r *Reporter) ScaleUp(req xfer.Request, resource, namespace, id string) xfer.Response {
|
||||
return xfer.ResponseError(r.client.ScaleUp(resource, namespace, id))
|
||||
func (r *Reporter) ScaleUp(req xfer.Request, namespace, id string) xfer.Response {
|
||||
return xfer.ResponseError(r.client.ScaleUp(report.Deployment, namespace, id))
|
||||
}
|
||||
|
||||
// ScaleDown is the control to scale up a deployment
|
||||
func (r *Reporter) ScaleDown(req xfer.Request, resource, namespace, id string) xfer.Response {
|
||||
return xfer.ResponseError(r.client.ScaleDown(resource, namespace, id))
|
||||
func (r *Reporter) ScaleDown(req xfer.Request, namespace, id string) xfer.Response {
|
||||
return xfer.ResponseError(r.client.ScaleDown(report.Deployment, namespace, id))
|
||||
}
|
||||
|
||||
func (r *Reporter) registerControls() {
|
||||
controls := map[string]xfer.ControlHandlerFunc{
|
||||
GetLogs: r.CapturePod(r.GetLogs),
|
||||
DeletePod: r.CapturePod(r.deletePod),
|
||||
ScaleUp: r.CaptureResource(r.ScaleUp),
|
||||
ScaleDown: r.CaptureResource(r.ScaleDown),
|
||||
ScaleUp: r.CaptureDeployment(r.ScaleUp),
|
||||
ScaleDown: r.CaptureDeployment(r.ScaleDown),
|
||||
}
|
||||
r.handlerRegistry.Batch(nil, controls)
|
||||
}
|
||||
|
||||
@@ -1,68 +0,0 @@
|
||||
package kubernetes
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
|
||||
"github.com/weaveworks/scope/report"
|
||||
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/apimachinery/pkg/labels"
|
||||
apiv1 "k8s.io/client-go/pkg/api/v1"
|
||||
apiv1beta1 "k8s.io/client-go/pkg/apis/extensions/v1beta1"
|
||||
)
|
||||
|
||||
// These constants are keys used in node metadata
|
||||
const (
|
||||
FullyLabeledReplicas = report.KubernetesFullyLabeledReplicas
|
||||
)
|
||||
|
||||
// ReplicaSet represents a Kubernetes replica set
|
||||
type ReplicaSet interface {
|
||||
Meta
|
||||
Selector() (labels.Selector, error)
|
||||
AddParent(topology, id string)
|
||||
GetNode(probeID string) report.Node
|
||||
}
|
||||
|
||||
type replicaSet struct {
|
||||
*apiv1beta1.ReplicaSet
|
||||
Meta
|
||||
parents report.Sets
|
||||
Node *apiv1.Node
|
||||
}
|
||||
|
||||
// NewReplicaSet creates a new ReplicaSet
|
||||
func NewReplicaSet(r *apiv1beta1.ReplicaSet) ReplicaSet {
|
||||
return &replicaSet{
|
||||
ReplicaSet: r,
|
||||
Meta: meta{r.ObjectMeta},
|
||||
parents: report.MakeSets(),
|
||||
}
|
||||
}
|
||||
|
||||
func (r *replicaSet) Selector() (labels.Selector, error) {
|
||||
selector, err := metav1.LabelSelectorAsSelector(r.Spec.Selector)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return selector, nil
|
||||
}
|
||||
|
||||
func (r *replicaSet) AddParent(topology, id string) {
|
||||
r.parents = r.parents.Add(topology, report.MakeStringSet(id))
|
||||
}
|
||||
|
||||
func (r *replicaSet) GetNode(probeID string) report.Node {
|
||||
// Spec.Replicas can be omitted, and the pointer will be nil. It defaults to 1.
|
||||
desiredReplicas := 1
|
||||
if r.Spec.Replicas != nil {
|
||||
desiredReplicas = int(*r.Spec.Replicas)
|
||||
}
|
||||
return r.MetaNode(report.MakeReplicaSetNodeID(r.UID())).WithLatests(map[string]string{
|
||||
ObservedGeneration: fmt.Sprint(r.Status.ObservedGeneration),
|
||||
Replicas: fmt.Sprint(r.Status.Replicas),
|
||||
DesiredReplicas: fmt.Sprint(desiredReplicas),
|
||||
FullyLabeledReplicas: fmt.Sprint(r.Status.FullyLabeledReplicas),
|
||||
report.ControlProbeID: probeID,
|
||||
}).WithParents(r.parents).WithLatestActiveControls(ScaleUp, ScaleDown)
|
||||
}
|
||||
@@ -1,56 +0,0 @@
|
||||
package kubernetes
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
|
||||
"github.com/weaveworks/scope/report"
|
||||
|
||||
"k8s.io/apimachinery/pkg/labels"
|
||||
apiv1 "k8s.io/client-go/pkg/api/v1"
|
||||
)
|
||||
|
||||
// ReplicationController represents a Kubernetes replication controller
|
||||
type ReplicationController interface {
|
||||
Meta
|
||||
Selector() (labels.Selector, error)
|
||||
AddParent(topology, id string)
|
||||
GetNode(probeID string) report.Node
|
||||
}
|
||||
|
||||
// replicationController implements both ReplicationController and ReplicaSet
|
||||
type replicationController struct {
|
||||
*apiv1.ReplicationController
|
||||
Meta
|
||||
parents report.Sets
|
||||
Node *apiv1.Node
|
||||
}
|
||||
|
||||
// NewReplicationController creates a new ReplicationController
|
||||
func NewReplicationController(r *apiv1.ReplicationController) ReplicationController {
|
||||
return &replicationController{
|
||||
ReplicationController: r,
|
||||
Meta: meta{r.ObjectMeta},
|
||||
parents: report.MakeSets(),
|
||||
}
|
||||
}
|
||||
|
||||
func (r *replicationController) Selector() (labels.Selector, error) {
|
||||
if r.Spec.Selector == nil {
|
||||
return labels.Nothing(), nil
|
||||
}
|
||||
return labels.SelectorFromSet(labels.Set(r.Spec.Selector)), nil
|
||||
}
|
||||
|
||||
func (r *replicationController) AddParent(topology, id string) {
|
||||
r.parents = r.parents.Add(topology, report.MakeStringSet(id))
|
||||
}
|
||||
|
||||
func (r *replicationController) GetNode(probeID string) report.Node {
|
||||
return r.MetaNode(report.MakeReplicaSetNodeID(r.UID())).WithLatests(map[string]string{
|
||||
ObservedGeneration: fmt.Sprint(r.Status.ObservedGeneration),
|
||||
Replicas: fmt.Sprint(r.Status.Replicas),
|
||||
DesiredReplicas: fmt.Sprint(r.Spec.Replicas),
|
||||
FullyLabeledReplicas: fmt.Sprint(r.Status.FullyLabeledReplicas),
|
||||
report.ControlProbeID: probeID,
|
||||
}).WithParents(r.parents).WithLatestActiveControls(ScaleUp, ScaleDown)
|
||||
}
|
||||
@@ -58,17 +58,7 @@ var (
|
||||
Strategy: {ID: Strategy, Label: "Strategy", From: report.FromLatest, Priority: 7},
|
||||
}
|
||||
|
||||
DeploymentMetricTemplates = ReplicaSetMetricTemplates
|
||||
|
||||
ReplicaSetMetadataTemplates = report.MetadataTemplates{
|
||||
Namespace: {ID: Namespace, Label: "Namespace", From: report.FromLatest, Priority: 2},
|
||||
Created: {ID: Created, Label: "Created", From: report.FromLatest, Datatype: report.DateTime, Priority: 3},
|
||||
ObservedGeneration: {ID: ObservedGeneration, Label: "Observed Gen.", From: report.FromLatest, Datatype: report.Number, Priority: 4},
|
||||
DesiredReplicas: {ID: DesiredReplicas, Label: "Desired Replicas", From: report.FromLatest, Datatype: report.Number, Priority: 5},
|
||||
report.Pod: {ID: report.Pod, Label: "# Pods", From: report.FromCounters, Datatype: report.Number, Priority: 6},
|
||||
}
|
||||
|
||||
ReplicaSetMetricTemplates = PodMetricTemplates
|
||||
DeploymentMetricTemplates = PodMetricTemplates
|
||||
|
||||
DaemonSetMetadataTemplates = report.MetadataTemplates{
|
||||
NodeType: {ID: NodeType, Label: "Type", From: report.FromLatest, Priority: 1},
|
||||
|
||||
@@ -145,12 +145,6 @@ func (c *mockClient) WalkCronJobs(f func(kubernetes.CronJob) error) error {
|
||||
func (c *mockClient) WalkDeployments(f func(kubernetes.Deployment) error) error {
|
||||
return nil
|
||||
}
|
||||
func (c *mockClient) WalkReplicaSets(f func(kubernetes.ReplicaSet) error) error {
|
||||
return nil
|
||||
}
|
||||
func (c *mockClient) WalkReplicationControllers(f func(kubernetes.ReplicationController) error) error {
|
||||
return nil
|
||||
}
|
||||
func (c *mockClient) WalkNamespaces(f func(kubernetes.NamespaceResource) error) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user