Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Update sink/source component observation logic. #217

Merged
merged 4 commits into from
Jun 28, 2021
Merged

Update sink/source component observation logic. #217

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
9163550
Update sink/source component observation logic.
MarvinCai Jun 23, 2021
18b7e1e
Address comments.
MarvinCai Jun 25, 2021
32e48a4
Fix checkstyle violation.
MarvinCai Jun 25, 2021
046411a
Fix golint error.
MarvinCai Jun 28, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
94 changes: 40 additions & 54 deletions controllers/sink.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,14 +32,10 @@ import (
)

func (r *SinkReconciler) ObserveSinkStatefulSet(ctx context.Context, req ctrl.Request, sink *v1alpha1.Sink) error {
condition, ok := sink.Status.Conditions[v1alpha1.StatefulSet]
if !ok {
sink.Status.Conditions[v1alpha1.StatefulSet] = v1alpha1.ResourceCondition{
Condition: v1alpha1.StatefulSetReady,
Status: metav1.ConditionFalse,
Action: v1alpha1.Create,
}
return nil
condition := v1alpha1.ResourceCondition{
Condition: v1alpha1.StatefulSetReady,
Status: metav1.ConditionFalse,
Action: v1alpha1.Create,
}

statefulSet := &appsv1.StatefulSet{}
Expand All @@ -49,33 +45,33 @@ func (r *SinkReconciler) ObserveSinkStatefulSet(ctx context.Context, req ctrl.Re
}, statefulSet)
if err != nil {
if errors.IsNotFound(err) {
r.Log.Info("sink is not ready yet...")
return nil
r.Log.Info("sink statefulset is not found...")
} else {
sink.Status.Conditions[v1alpha1.StatefulSet] = condition
return err
MarvinCai marked this conversation as resolved.
Show resolved Hide resolved
}
return err
} else {
MarvinCai marked this conversation as resolved.
Show resolved Hide resolved
// statefulset created, waiting it to be ready
condition.Action = v1alpha1.Wait
}

selector, err := metav1.LabelSelectorAsSelector(statefulSet.Spec.Selector)
if err != nil {
r.Log.Error(err, "error retrieving statefulSet selector")
return err
} else {
MarvinCai marked this conversation as resolved.
Show resolved Hide resolved
sink.Status.Selector = selector.String()
}
sink.Status.Selector = selector.String()

if *statefulSet.Spec.Replicas != *sink.Spec.Replicas {
condition.Status = metav1.ConditionFalse
condition.Action = v1alpha1.Update
sink.Status.Conditions[v1alpha1.StatefulSet] = condition
return nil
}

if statefulSet.Status.ReadyReplicas == *sink.Spec.Replicas {
condition.Action = v1alpha1.NoAction
condition.Status = metav1.ConditionTrue
} else {
condition.Action = v1alpha1.Wait
}
sink.Status.Replicas = *statefulSet.Spec.Replicas

sink.Status.Replicas = statefulSet.Status.ReadyReplicas
MarvinCai marked this conversation as resolved.
Show resolved Hide resolved
sink.Status.Conditions[v1alpha1.StatefulSet] = condition

return nil
Expand All @@ -101,26 +97,18 @@ func (r *SinkReconciler) ApplySinkStatefulSet(ctx context.Context, req ctrl.Requ
r.Log.Error(err, "failed to update the sink statefulSet")
return err
}
case v1alpha1.Wait:
case v1alpha1.Wait, v1alpha1.NoAction:
// do nothing
}

return nil
}

func (r *SinkReconciler) ObserveSinkService(ctx context.Context, req ctrl.Request, sink *v1alpha1.Sink) error {
condition, ok := sink.Status.Conditions[v1alpha1.Service]
if !ok {
sink.Status.Conditions[v1alpha1.Service] = v1alpha1.ResourceCondition{
Condition: v1alpha1.ServiceReady,
Status: metav1.ConditionFalse,
Action: v1alpha1.Create,
}
return nil
}

if condition.Status == metav1.ConditionTrue {
return nil
condition := v1alpha1.ResourceCondition{
Condition: v1alpha1.ServiceReady,
Status: metav1.ConditionFalse,
Action: v1alpha1.Create,
}

svc := &corev1.Service{}
Expand All @@ -129,13 +117,16 @@ func (r *SinkReconciler) ObserveSinkService(ctx context.Context, req ctrl.Reques
if err != nil {
if errors.IsNotFound(err) {
r.Log.Info("service is not created...", "Name", sink.Name)
return nil
} else {
sink.Status.Conditions[v1alpha1.Service] = condition
return err
}
return err
} else {
// service object doesn't have status, so once it's created just consider it's ready
condition.Action = v1alpha1.NoAction
condition.Status = metav1.ConditionTrue
}

condition.Action = v1alpha1.NoAction
condition.Status = metav1.ConditionTrue
sink.Status.Conditions[v1alpha1.Service] = condition
return nil
}
Expand All @@ -154,7 +145,7 @@ func (r *SinkReconciler) ApplySinkService(ctx context.Context, req ctrl.Request,
r.Log.Error(err, "failed to expose service for sink", "name", sink.Name)
return err
}
case v1alpha1.Wait:
case v1alpha1.Wait, v1alpha1.NoAction:
// do nothing
}

Expand All @@ -167,18 +158,10 @@ func (r *SinkReconciler) ObserveSinkHPA(ctx context.Context, req ctrl.Request, s
return nil
}

condition, ok := sink.Status.Conditions[v1alpha1.HPA]
if !ok {
sink.Status.Conditions[v1alpha1.HPA] = v1alpha1.ResourceCondition{
Condition: v1alpha1.HPAReady,
Status: metav1.ConditionFalse,
Action: v1alpha1.Create,
}
return nil
}

if condition.Status == metav1.ConditionTrue {
return nil
condition := v1alpha1.ResourceCondition{
Condition: v1alpha1.HPAReady,
Status: metav1.ConditionFalse,
Action: v1alpha1.Create,
}

hpa := &autov1.HorizontalPodAutoscaler{}
Expand All @@ -187,13 +170,16 @@ func (r *SinkReconciler) ObserveSinkHPA(ctx context.Context, req ctrl.Request, s
if err != nil {
if errors.IsNotFound(err) {
r.Log.Info("hpa is not created for sink...", "name", sink.Name)
return nil
} else {
sink.Status.Conditions[v1alpha1.HPA] = condition
return err
}
return err
} else {
// HPA's status doesn't show its readiness, , so once it's created just consider it's ready
MarvinCai marked this conversation as resolved.
Show resolved Hide resolved
condition.Action = v1alpha1.NoAction
condition.Status = metav1.ConditionTrue
}

condition.Action = v1alpha1.NoAction
condition.Status = metav1.ConditionTrue
sink.Status.Conditions[v1alpha1.HPA] = condition
return nil
}
Expand All @@ -217,7 +203,7 @@ func (r *SinkReconciler) ApplySinkHPA(ctx context.Context, req ctrl.Request, sin
r.Log.Error(err, "failed to create pod autoscaler for sink", "name", sink.Name)
return err
}
case v1alpha1.Wait:
case v1alpha1.Wait, v1alpha1.NoAction:
// do nothing
}

Expand Down
94 changes: 40 additions & 54 deletions controllers/source.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,14 +33,10 @@ import (

func (r *SourceReconciler) ObserveSourceStatefulSet(ctx context.Context, req ctrl.Request,
source *v1alpha1.Source) error {
condition, ok := source.Status.Conditions[v1alpha1.StatefulSet]
if !ok {
source.Status.Conditions[v1alpha1.StatefulSet] = v1alpha1.ResourceCondition{
Condition: v1alpha1.StatefulSetReady,
Status: metav1.ConditionFalse,
Action: v1alpha1.Create,
}
return nil
condition := v1alpha1.ResourceCondition{
Condition: v1alpha1.StatefulSetReady,
Status: metav1.ConditionFalse,
Action: v1alpha1.Create,
}

statefulSet := &appsv1.StatefulSet{}
Expand All @@ -50,33 +46,33 @@ func (r *SourceReconciler) ObserveSourceStatefulSet(ctx context.Context, req ctr
}, statefulSet)
if err != nil {
if errors.IsNotFound(err) {
r.Log.Info("source is not ready yet...")
return nil
r.Log.Info("sink statefulset is not found...")
} else {
source.Status.Conditions[v1alpha1.StatefulSet] = condition
return err
}
return err
} else {
// statefulset created, waiting it to be ready
condition.Action = v1alpha1.Wait
}

selector, err := metav1.LabelSelectorAsSelector(statefulSet.Spec.Selector)
if err != nil {
r.Log.Error(err, "error retrieving statefulSet selector")
return err
} else {
source.Status.Selector = selector.String()
}
source.Status.Selector = selector.String()

if *statefulSet.Spec.Replicas != *source.Spec.Replicas {
condition.Status = metav1.ConditionFalse
condition.Action = v1alpha1.Update
source.Status.Conditions[v1alpha1.StatefulSet] = condition
return nil
}

if statefulSet.Status.ReadyReplicas == *source.Spec.Replicas {
condition.Action = v1alpha1.NoAction
condition.Status = metav1.ConditionTrue
} else {
condition.Action = v1alpha1.Wait
}
source.Status.Replicas = *statefulSet.Spec.Replicas

source.Status.Replicas = statefulSet.Status.ReadyReplicas
source.Status.Conditions[v1alpha1.StatefulSet] = condition

return nil
Expand All @@ -103,26 +99,18 @@ func (r *SourceReconciler) ApplySourceStatefulSet(ctx context.Context, req ctrl.
r.Log.Error(err, "failed to update the source statefulSet")
return err
}
case v1alpha1.Wait:
case v1alpha1.Wait, v1alpha1.NoAction:
// do nothing
}

return nil
}

func (r *SourceReconciler) ObserveSourceService(ctx context.Context, req ctrl.Request, source *v1alpha1.Source) error {
condition, ok := source.Status.Conditions[v1alpha1.Service]
if !ok {
source.Status.Conditions[v1alpha1.Service] = v1alpha1.ResourceCondition{
Condition: v1alpha1.ServiceReady,
Status: metav1.ConditionFalse,
Action: v1alpha1.Create,
}
return nil
}

if condition.Status == metav1.ConditionTrue {
return nil
condition := v1alpha1.ResourceCondition{
Condition: v1alpha1.ServiceReady,
Status: metav1.ConditionFalse,
Action: v1alpha1.Create,
}

svc := &corev1.Service{}
Expand All @@ -131,13 +119,16 @@ func (r *SourceReconciler) ObserveSourceService(ctx context.Context, req ctrl.Re
if err != nil {
if errors.IsNotFound(err) {
r.Log.Info("service is not created...", "Name", source.Name)
return nil
} else {
source.Status.Conditions[v1alpha1.Service] = condition
return err
}
return err
} else {
// service object doesn't have status, so once it's created just consider it's ready
condition.Action = v1alpha1.NoAction
condition.Status = metav1.ConditionTrue
}

condition.Action = v1alpha1.NoAction
condition.Status = metav1.ConditionTrue
source.Status.Conditions[v1alpha1.Service] = condition
return nil
}
Expand All @@ -156,7 +147,7 @@ func (r *SourceReconciler) ApplySourceService(ctx context.Context, req ctrl.Requ
r.Log.Error(err, "failed to expose service for source", "name", source.Name)
return err
}
case v1alpha1.Wait:
case v1alpha1.Wait, v1alpha1.NoAction:
// do nothing
}

Expand All @@ -169,18 +160,10 @@ func (r *SourceReconciler) ObserveSourceHPA(ctx context.Context, req ctrl.Reques
return nil
}

condition, ok := source.Status.Conditions[v1alpha1.HPA]
if !ok {
source.Status.Conditions[v1alpha1.HPA] = v1alpha1.ResourceCondition{
Condition: v1alpha1.HPAReady,
Status: metav1.ConditionFalse,
Action: v1alpha1.Create,
}
return nil
}

if condition.Status == metav1.ConditionTrue {
return nil
condition := v1alpha1.ResourceCondition{
Condition: v1alpha1.HPAReady,
Status: metav1.ConditionFalse,
Action: v1alpha1.Create,
}

hpa := &autov1.HorizontalPodAutoscaler{}
Expand All @@ -189,13 +172,16 @@ func (r *SourceReconciler) ObserveSourceHPA(ctx context.Context, req ctrl.Reques
if err != nil {
if errors.IsNotFound(err) {
r.Log.Info("hpa is not created for source...", "name", source.Name)
return nil
} else {
source.Status.Conditions[v1alpha1.HPA] = condition
return err
}
return err
} else {
// HPA's status doesn't show its readiness, , so once it's created just consider it's ready
condition.Action = v1alpha1.NoAction
condition.Status = metav1.ConditionTrue
}

condition.Action = v1alpha1.NoAction
condition.Status = metav1.ConditionTrue
source.Status.Conditions[v1alpha1.HPA] = condition
return nil
}
Expand All @@ -219,7 +205,7 @@ func (r *SourceReconciler) ApplySourceHPA(ctx context.Context, req ctrl.Request,
r.Log.Error(err, "failed to create pod autoscaler for source", "name", source.Name)
return err
}
case v1alpha1.Wait:
case v1alpha1.Wait, v1alpha1.NoAction:
// do nothing
}

Expand Down