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

SKS use one service #3711

Merged
merged 9 commits into from
Apr 10, 2019
+50 −164
Merged

SKS use one service #3711

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
86ac254
Use generic pkg/resources helpers for endpoint counting
vagababov Apr 9, 2019
b7487ba
Move SKS helpers to a shared place.
vagababov Apr 9, 2019
0d953df
some transient state
vagababov Apr 9, 2019
af8f308
remove public service and endpoint reconciliation
vagababov Apr 9, 2019
f2186e6
remove public service and endpoint reconciliation
vagababov Apr 9, 2019
d3e3fcc
formatting unification
vagababov Apr 10, 2019
5756581
merge
vagababov Apr 10, 2019
95e01d5
master merge
vagababov Apr 10, 2019
bd9687b
Merge branch 'master' into sks-use-one-service
vagababov Apr 10, 2019
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
40 changes: 28 additions & 12 deletions pkg/reconciler/v1alpha1/serverlessservice/serverlessservice.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@ import (

"github.com/davecgh/go-spew/spew"
"github.com/google/go-cmp/cmp"
perrors "github.com/pkg/errors"

"github.com/knative/pkg/controller"
"github.com/knative/pkg/logging"
Expand Down Expand Up @@ -166,11 +167,9 @@ func (r *reconciler) reconcile(ctx context.Context, sks *netv1alpha1.ServerlessS
sks.SetDefaults(ctx)
sks.Status.InitializeConditions()

// TODO(#1997): implement: proxy mode, activator probing and positive handoff.
// TODO(#1997): implement: public service, proxy mode, activator probing and positive handoff.
for i, fn := range []func(context.Context, *netv1alpha1.ServerlessService) error{
r.reconcilePrivateService, // First make sure our data source is setup.
r.reconcilePublicService, // Make sure the service is setup.
r.reconcilePublicEndpoints, // Now populate endpoints, if there are healthy ones.
r.reconcilePrivateService, // First make sure our data source is setup.
} {
if err := fn(ctx, sks); err != nil {
logger.Debugw(fmt.Sprintf("%d: reconcile failed", i), zap.Error(err))
Expand Down Expand Up @@ -295,21 +294,21 @@ func (r *reconciler) reconcilePrivateService(ctx context.Context, sks *netv1alph
sn := names.PrivateService(sks)
svc, err := r.serviceLister.Services(sks.Namespace).Get(sn)
if errors.IsNotFound(err) {
logger.Infof("K8s service %q does not exist; creating.", sn)
logger.Infof("K8s service %s does not exist; creating.", sn)
sks.Status.MarkEndpointsNotReady("CreatingPrivateService")
svc = resources.MakePrivateService(sks)
_, err := r.KubeClientSet.CoreV1().Services(sks.Namespace).Create(svc)
if err != nil {
logger.Errorw(fmt.Sprintf("Error creating K8s Service %s: ", sn), zap.Error(err))
logger.Errorw(fmt.Sprint("Error creating K8s Service:", sn), zap.Error(err))
return err
}
logger.Infof("Created K8s service: %q", sn)
logger.Infof("Created K8s service: %s", sn)
} else if err != nil {
logger.Errorw(fmt.Sprintf("Error getting K8s Service %s: ", sn), zap.Error(err))
logger.Errorw(fmt.Sprint("Error getting K8s Service:", sn), zap.Error(err))
return err
} else if !metav1.IsControlledBy(svc, sks) {
sks.Status.MarkEndpointsNotOwned("Service", sn)
return fmt.Errorf("SKS: %q does not own Service: %q", sks.Name, sn)
return fmt.Errorf("SKS: %s does not own Service: %s", sks.Name, sn)
}
tmpl := resources.MakePrivateService(sks)
want := svc.DeepCopy()
Expand All @@ -319,13 +318,30 @@ func (r *reconciler) reconcilePrivateService(ctx context.Context, sks *netv1alph

if !equality.Semantic.DeepEqual(svc.Spec, want.Spec) {
sks.Status.MarkEndpointsNotReady("UpdatingPrivateService")
logger.Infof("Private K8s Service changed; reconciling: %s", sn)
logger.Info("Private K8s Service changed; reconciling:", sn)
if _, err = r.KubeClientSet.CoreV1().Services(sks.Namespace).Update(want); err != nil {
logger.Errorw(fmt.Sprintf("Error updating private K8s Service %s: ", sn), zap.Error(err))
logger.Errorw(fmt.Sprint("Error updating private K8s Service:", sn), zap.Error(err))
return err
}
}

// TODO(#1997): temporarily we have one service since istio cannot handle our load.
// So they are smudged together. In the end this has to go.
eps, err := presources.FetchReadyAddressCount(r.endpointsLister, sks.Namespace, sn)
switch {
case err != nil:
return perrors.Wrapf(err, "error fetching endpoints %s/%s", sks.Namespace, sn)
case eps > 0:
logger.Infof("Endpoints %s/%s has %d ready endpoints", sks.Namespace, sn, eps)
sks.Status.MarkEndpointsReady()
default:
logger.Infof("Endpoints %s/%s has no ready endpoints", sks.Namespace, sn)
sks.Status.MarkEndpointsNotReady("NoHealthyBackends")
}
sks.Status.ServiceName = sn
// End TODO.

sks.Status.PrivateServiceName = sn
logger.Debugf("Done reconciling private K8s service %s", sn)
logger.Debug("Done reconciling private K8s service", sn)
return nil
}
174 changes: 22 additions & 152 deletions pkg/reconciler/v1alpha1/serverlessservice/serverlessservice_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@ import (
informers "github.com/knative/serving/pkg/client/informers/externalversions"
rpkg "github.com/knative/serving/pkg/reconciler"
"github.com/knative/serving/pkg/reconciler/v1alpha1/serverlessservice/resources"
"github.com/knative/serving/pkg/reconciler/v1alpha1/serverlessservice/resources/names"

corev1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
Expand Down Expand Up @@ -80,50 +81,24 @@ func TestReconcile(t *testing.T) {
Name: "steady state",
Key: "steady/state",
Objects: []runtime.Object{
SKS("steady", "state", markHappy, WithPubService, WithPrivateService),
SKS("steady", "state", markHappy, withTempPubService, WithPrivateService),
svcpub("steady", "state"),
svcpriv("steady", "state"),
endpointspub("steady", "state", WithSubsets),
endpointspriv("steady", "state", WithSubsets),
},
}, {
Name: "pod change",
Key: "pod/change",
Objects: []runtime.Object{
SKS("pod", "change", markHappy, WithPubService, WithPrivateService),
svcpub("pod", "change"),
svcpriv("pod", "change"),
endpointspub("pod", "change", WithSubsets),
endpointspriv("pod", "change", withOtherSubsets),
},
WantUpdates: []clientgotesting.UpdateActionImpl{{
Object: endpointspub("pod", "change", withOtherSubsets),
}},
}, {
Name: "user changes priv svc",
Key: "private/svc-change",
Objects: []runtime.Object{
SKS("private", "svc-change", markHappy, WithPubService, WithPrivateService),
SKS("private", "svc-change", markHappy, withTempPubService, WithPrivateService),
svcpub("private", "svc-change"),
svcpriv("private", "svc-change", withTimeSelector),
endpointspub("private", "svc-change", withOtherSubsets),
endpointspriv("private", "svc-change", WithSubsets),
},
WantUpdates: []clientgotesting.UpdateActionImpl{{
Object: svcpriv("private", "svc-change")}, {
Object: endpointspub("private", "svc-change", WithSubsets)}},
}, {
Name: "user changes public svc",
Key: "public/svc-change",
Objects: []runtime.Object{
SKS("public", "svc-change", markHappy, WithPubService, WithPrivateService),
svcpub("public", "svc-change", withTimeSelector),
svcpriv("public", "svc-change"),
endpointspub("public", "svc-change", WithSubsets),
endpointspriv("public", "svc-change", WithSubsets),
},
WantUpdates: []clientgotesting.UpdateActionImpl{{
Object: svcpub("public", "svc-change"),
Object: svcpriv("private", "svc-change"),
}},
}, {
Name: "OnCreate-deployment-exists",
Expand All @@ -135,11 +110,9 @@ func TestReconcile(t *testing.T) {
},
WantCreates: []metav1.Object{
svcpriv("on", "cde"),
svcpub("on", "cde"),
endpointspub("on", "cde", WithSubsets),
},
WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
Object: SKS("on", "cde", markHappy, WithPubService, WithPrivateService),
Object: SKS("on", "cde", markHappy, withTempPubService, WithPrivateService),
}},
WantEvents: []string{
Eventf(corev1.EventTypeNormal, "Updated", `Successfully updated ServerlessService "on/cde"`),
Expand All @@ -153,11 +126,9 @@ func TestReconcile(t *testing.T) {
},
WantCreates: []metav1.Object{
svcpriv("on", "cneps"),
svcpub("on", "cneps"),
endpointspub("on", "cneps"),
},
WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
Object: SKS("on", "cneps", markNoEndpoints, WithPubService, WithPrivateService),
Object: SKS("on", "cneps", markNoEndpoints, withTempPubService, WithPrivateService),
}},
WantEvents: []string{
Eventf(corev1.EventTypeNormal, "Updated", `Successfully updated ServerlessService "on/cneps"`),
Expand All @@ -179,49 +150,12 @@ func TestReconcile(t *testing.T) {
WantEvents: []string{
Eventf(corev1.EventTypeWarning, "UpdateFailed", "InternalError: inducing failure for create services"),
},
}, {
Name: "svc-fail-pub",
Key: "svc/fail2",
WantErr: true,
Objects: []runtime.Object{
SKS("svc", "fail2"),
svcpriv("svc", "fail2"),
endpointspriv("svc", "fail2"),
},
WithReactors: []clientgotesting.ReactionFunc{
InduceFailure("create", "services"),
},
WantCreates: []metav1.Object{
svcpub("svc", "fail2"),
},
WantEvents: []string{
Eventf(corev1.EventTypeWarning, "UpdateFailed", "InternalError: inducing failure for create services"),
},
}, {
Name: "eps-fail-pub",
Key: "eps/fail3",
WantErr: true,
Objects: []runtime.Object{
SKS("eps", "fail3"),
svcpriv("eps", "fail3"),
endpointspriv("eps", "fail3"),
},
WithReactors: []clientgotesting.ReactionFunc{
InduceFailure("create", "endpoints"),
},
WantCreates: []metav1.Object{
svcpub("eps", "fail3"),
endpointspub("eps", "fail3"),
},
WantEvents: []string{
Eventf(corev1.EventTypeWarning, "UpdateFailed", "InternalError: inducing failure for create endpoints"),
},
}, {
Name: "update-sks-fail",
Key: "update-sks/fail4",
WantErr: true,
Objects: []runtime.Object{
SKS("update-sks", "fail4", WithPubService, WithPrivateService),
SKS("update-sks", "fail4", withTempPubService, WithPrivateService),
svcpub("update-sks", "fail4"),
svcpriv("update-sks", "fail4"),
endpointspub("update-sks", "fail4", WithSubsets),
Expand All @@ -232,79 +166,31 @@ func TestReconcile(t *testing.T) {
},
// We still record update, but it fails.
WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
Object: SKS("update-sks", "fail4", markHappy, WithPubService, WithPrivateService),
Object: SKS("update-sks", "fail4", markHappy, withTempPubService, WithPrivateService),
}},
WantEvents: []string{
Eventf(corev1.EventTypeWarning, "UpdateFailed", "Failed to update status: inducing failure for update serverlessservices"),
},
}, {
Name: "ronin-pub-service/fail5",
Key: "ronin-pub-service/fail5",
WantErr: true,
Objects: []runtime.Object{
SKS("ronin-pub-service", "fail5", WithPubService, WithPrivateService),
svcpub("ronin-pub-service", "fail5", WithK8sSvcOwnersRemoved),
svcpriv("ronin-pub-service", "fail5"),
endpointspub("ronin-pub-service", "fail5", WithSubsets),
endpointspriv("ronin-pub-service", "fail5", WithSubsets),
},
WantEvents: []string{
Eventf(corev1.EventTypeWarning, "UpdateFailed", `InternalError: SKS: "fail5" does not own Service: "fail5-pub"`),
},
}, {
Name: "ronin-priv-service/fail6",
Key: "ronin-priv-service/fail6",
WantErr: true,
Objects: []runtime.Object{
SKS("ronin-priv-service", "fail6", WithPubService, WithPrivateService),
svcpub("ronin-priv-service", "fail6"),
svcpriv("ronin-priv-service", "fail6", WithK8sSvcOwnersRemoved),
endpointspub("ronin-priv-service", "fail6", WithSubsets),
endpointspriv("ronin-priv-service", "fail6", WithSubsets),
},
WantEvents: []string{
Eventf(corev1.EventTypeWarning, "UpdateFailed", `InternalError: SKS: "fail6" does not own Service: "fail6-priv"`),
},
}, {
Name: "ronin-pub-eps/fail7",
Key: "ronin-pub-eps/fail7",
Name: "ronin-priv-service/fail5",
Key: "ronin-priv-service/fail5",
WantErr: true,
Objects: []runtime.Object{
SKS("ronin-pub-eps", "fail7", WithPubService),
svcpub("ronin-pub-eps", "fail7"),
svcpriv("ronin-pub-eps", "fail7"),
endpointspub("ronin-pub-eps", "fail7", WithSubsets, WithEndpointsOwnersRemoved),
endpointspriv("ronin-pub-eps", "fail7", WithSubsets),
},
WantEvents: []string{
Eventf(corev1.EventTypeWarning, "UpdateFailed", `InternalError: SKS: "fail7" does not own Endpoints: "fail7-pub"`),
},
}, {
Name: "update-svc-fail-pub",
Key: "update-svc/fail8",
WantErr: true,
Objects: []runtime.Object{
SKS("update-svc", "fail8", WithPubService),
svcpub("update-svc", "fail8", withTimeSelector),
svcpriv("update-svc", "fail8"),
endpointspub("update-svc", "fail8", WithSubsets),
endpointspriv("update-svc", "fail8", WithSubsets),
},
WithReactors: []clientgotesting.ReactionFunc{
InduceFailure("update", "services"),
SKS("ronin-priv-service", "fail5", withTempPubService, WithPrivateService),
svcpub("ronin-priv-service", "fail5"),
svcpriv("ronin-priv-service", "fail5", WithK8sSvcOwnersRemoved),
endpointspub("ronin-priv-service", "fail5", WithSubsets),
endpointspriv("ronin-priv-service", "fail5", WithSubsets),
},
WantUpdates: []clientgotesting.UpdateActionImpl{{
Object: svcpub("update-svc", "fail8"),
}},
WantEvents: []string{
Eventf(corev1.EventTypeWarning, "UpdateFailed", "InternalError: inducing failure for update services"),
Eventf(corev1.EventTypeWarning, "UpdateFailed", `InternalError: SKS: fail5 does not own Service: fail5-priv`),
},
}, {
Name: "update-svc-fail-priv",
Key: "update-svc/fail9",
WantErr: true,
Objects: []runtime.Object{
SKS("update-svc", "fail9", WithPubService, WithPrivateService),
SKS("update-svc", "fail9", withTempPubService, WithPrivateService),
svcpub("update-svc", "fail9"),
svcpriv("update-svc", "fail9", withTimeSelector),
endpointspub("update-svc", "fail9"),
Expand All @@ -320,27 +206,6 @@ func TestReconcile(t *testing.T) {
WantEvents: []string{
Eventf(corev1.EventTypeWarning, "UpdateFailed", "InternalError: inducing failure for update services"),
},
}, {
Name: "update-eps-fail",
Key: "update-eps/failA",
WantErr: true,
Objects: []runtime.Object{
SKS("update-eps", "failA", WithPubService, WithPrivateService),
svcpub("update-eps", "failA"),
svcpriv("update-eps", "failA"),
endpointspub("update-eps", "failA"),
endpointspriv("update-eps", "failA", WithSubsets),
},
WithReactors: []clientgotesting.ReactionFunc{
InduceFailure("update", "endpoints"),
},
WantUpdates: []clientgotesting.UpdateActionImpl{{
Object: endpointspub("update-eps", "failA", WithSubsets), // The attempted update.
}},

WantEvents: []string{
Eventf(corev1.EventTypeWarning, "UpdateFailed", "InternalError: inducing failure for update endpoints"),
},
}}

defer ClearAllLoggers()
Expand Down Expand Up @@ -415,3 +280,8 @@ func endpointspub(namespace, name string, eo ...EndpointsOption) *corev1.Endpoin
func withTimeSelector(svc *corev1.Service) {
svc.Spec.Selector = map[string]string{"pod-x": fmt.Sprintf("a-%d", time.Now().UnixNano())}
}

// TODO(vagababov): temp while we don't create separate public service.
func withTempPubService(sks *nv1a1.ServerlessService) {
sks.Status.ServiceName = names.PrivateService(sks)
}