Skip to content

Commit 79c801e

Browse files
AdilFayyazAdilFayyaz
authored
feat: implement InternalAppService for Knative app lifecycle (#7175)
* add: dep Signed-off-by: M. Adil Fayyaz <62440954+AdilFayyaz@users.noreply.github.com> * restructure Signed-off-by: M. Adil Fayyaz <62440954+AdilFayyaz@users.noreply.github.com> * wip Signed-off-by: M. Adil Fayyaz <62440954+AdilFayyaz@users.noreply.github.com> * restruct Signed-off-by: M. Adil Fayyaz <62440954+AdilFayyaz@users.noreply.github.com> * fix Signed-off-by: M. Adil Fayyaz <62440954+AdilFayyaz@users.noreply.github.com> * move config Signed-off-by: M. Adil Fayyaz <62440954+AdilFayyaz@users.noreply.github.com> * remove db.go Signed-off-by: M. Adil Fayyaz <62440954+AdilFayyaz@users.noreply.github.com> * address comments Signed-off-by: M. Adil Fayyaz <62440954+AdilFayyaz@users.noreply.github.com> * impl: internal service Signed-off-by: M. Adil Fayyaz <62440954+AdilFayyaz@users.noreply.github.com> * add tests Signed-off-by: M. Adil Fayyaz <62440954+AdilFayyaz@users.noreply.github.com> * address comments Signed-off-by: M. Adil Fayyaz <62440954+AdilFayyaz@users.noreply.github.com> --------- Signed-off-by: M. Adil Fayyaz <62440954+AdilFayyaz@users.noreply.github.com>
1 parent 17dd7d9 commit 79c801e

6 files changed

Lines changed: 748 additions & 24 deletions

File tree

‎app/config/config.go‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,10 @@ type AppConfig struct {
77
// Enabled controls whether the app deployment controller is started.
88
Enabled bool `json:"enabled" pflag:",Enable app deployment controller"`
99

10+
// BaseDomain is the base domain used to generate public URLs for apps.
11+
// Apps are exposed at "{name}-{project}-{domain}.{base_domain}".
12+
BaseDomain string `json:"baseDomain" pflag:",Base domain for app public URLs"`
13+
1014
// DefaultRequestTimeout is the request timeout applied to apps that don't specify one.
1115
DefaultRequestTimeout time.Duration `json:"defaultRequestTimeout" pflag:",Default request timeout for apps"`
1216

‎app/internal/k8s/app_client.go‎

Lines changed: 42 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -52,8 +52,11 @@ type AppK8sClientInterface interface {
5252
// Returns a not-found error (checkable with k8serrors.IsNotFound) if the KService does not exist.
5353
GetStatus(ctx context.Context, appID *flyteapp.Identifier) (*flyteapp.Status, error)
5454

55-
// List returns all apps (spec + live status) for the given project/domain scope.
56-
List(ctx context.Context, project, domain string) ([]*flyteapp.App, error)
55+
// List returns apps for the given project/domain scope with optional pagination.
56+
// If appName is non-empty, only the app with that name is returned.
57+
// limit=0 means no limit. token is the K8s continue token from a previous call.
58+
// Returns the apps, the continue token for the next page (empty if last page), and any error.
59+
List(ctx context.Context, project, domain, appName string, limit uint32, token string) ([]*flyteapp.App, string, error)
5760

5861
// Delete removes the KService CRD entirely. The app must be re-created from scratch.
5962
// Use Stop to scale to zero while preserving the KService.
@@ -66,8 +69,9 @@ type AppK8sClientInterface interface {
6669
DeleteReplica(ctx context.Context, replicaID *flyteapp.ReplicaIdentifier) error
6770

6871
// Watch returns a channel of WatchResponse events for KServices matching the
69-
// given project/domain scope. The channel is closed when ctx is cancelled.
70-
Watch(ctx context.Context, project, domain string) (<-chan *flyteapp.WatchResponse, error)
72+
// given project/domain scope. If appName is non-empty, only events for that
73+
// specific app are returned. The channel is closed when ctx is cancelled.
74+
Watch(ctx context.Context, project, domain, appName string) (<-chan *flyteapp.WatchResponse, error)
7175
}
7276

7377
// AppK8sClient implements AppK8sClientInterface using controller-runtime.
@@ -169,13 +173,20 @@ func (c *AppK8sClient) Delete(ctx context.Context, appID *flyteapp.Identifier) e
169173
}
170174

171175
// Watch returns a channel of WatchResponse events for KServices in the given
172-
// project/domain scope. The channel is closed when ctx is cancelled or the
176+
// project/domain scope. If appName is non-empty, only events for that specific
177+
// app are returned. The channel is closed when ctx is cancelled or the
173178
// underlying watch terminates.
174-
func (c *AppK8sClient) Watch(ctx context.Context, project, domain string) (<-chan *flyteapp.WatchResponse, error) {
179+
func (c *AppK8sClient) Watch(ctx context.Context, project, domain, appName string) (<-chan *flyteapp.WatchResponse, error) {
175180
ns := appNamespace(project, domain)
181+
182+
labels := map[string]string{labelAppManaged: "true"}
183+
if appName != "" {
184+
labels[labelAppName] = strings.ToLower(appName)
185+
}
186+
176187
watcher, err := c.k8sClient.Watch(ctx, &servingv1.ServiceList{},
177188
client.InNamespace(ns),
178-
client.MatchingLabels{labelAppManaged: "true"},
189+
client.MatchingLabels(labels),
179190
)
180191
if err != nil {
181192
return nil, fmt.Errorf("failed to start KService watch in namespace %s: %w", ns, err)
@@ -258,16 +269,28 @@ func (c *AppK8sClient) GetStatus(ctx context.Context, appID *flyteapp.Identifier
258269
return c.kserviceToStatus(ctx, ksvc), nil
259270
}
260271

261-
// List returns all apps for the given project/domain by listing KServices in the
262-
// project/domain namespace.
263-
func (c *AppK8sClient) List(ctx context.Context, project, domain string) ([]*flyteapp.App, error) {
272+
// List returns apps for the given project/domain scope with optional pagination.
273+
func (c *AppK8sClient) List(ctx context.Context, project, domain, appName string, limit uint32, token string) ([]*flyteapp.App, string, error) {
264274
ns := appNamespace(project, domain)
265-
list := &servingv1.ServiceList{}
266-
if err := c.k8sClient.List(ctx, list,
275+
276+
matchLabels := client.MatchingLabels{labelAppManaged: "true"}
277+
if appName != "" {
278+
matchLabels[labelAppName] = strings.ToLower(appName)
279+
}
280+
listOpts := []client.ListOption{
267281
client.InNamespace(ns),
268-
client.MatchingLabels{labelAppManaged: "true"},
269-
); err != nil {
270-
return nil, fmt.Errorf("failed to list KServices for %s/%s: %w", project, domain, err)
282+
matchLabels,
283+
}
284+
if limit > 0 {
285+
listOpts = append(listOpts, client.Limit(int64(limit)))
286+
}
287+
if token != "" {
288+
listOpts = append(listOpts, client.Continue(token))
289+
}
290+
291+
list := &servingv1.ServiceList{}
292+
if err := c.k8sClient.List(ctx, list, listOpts...); err != nil {
293+
return nil, "", fmt.Errorf("failed to list KServices for %s/%s: %w", project, domain, err)
271294
}
272295

273296
apps := make([]*flyteapp.App, 0, len(list.Items))
@@ -279,7 +302,7 @@ func (c *AppK8sClient) List(ctx context.Context, project, domain string) ([]*fly
279302
}
280303
apps = append(apps, a)
281304
}
282-
return apps, nil
305+
return apps, list.Continue, nil
283306
}
284307

285308
// --- Helpers ---
@@ -302,7 +325,7 @@ func kserviceName(id *flyteapp.Identifier) string {
302325

303326
// specSHA computes a SHA256 digest of the serialized App Spec proto.
304327
func specSHA(spec *flyteapp.Spec) (string, error) {
305-
b, err := proto.MarshalOptions{Deterministic: true}.Marshal(spec)
328+
b, err := proto.Marshal(spec)
306329
if err != nil {
307330
return "", fmt.Errorf("failed to marshal spec: %w", err)
308331
}
@@ -358,10 +381,6 @@ func (c *AppK8sClient) buildKService(app *flyteapp.App) (*servingv1.Service, err
358381
Template: servingv1.RevisionTemplateSpec{
359382
ObjectMeta: metav1.ObjectMeta{
360383
Annotations: templateAnnotations,
361-
Labels: map[string]string{
362-
labelAppManaged: "true",
363-
labelAppName: appID.GetName(),
364-
},
365384
},
366385
Spec: servingv1.RevisionSpec{
367386
PodSpec: podSpec,
@@ -467,8 +486,8 @@ func (c *AppK8sClient) kserviceToStatus(ctx context.Context, ksvc *servingv1.Ser
467486
phase = flyteapp.Status_DEPLOYMENT_STATUS_ACTIVE
468487
case ksvc.IsFailed():
469488
phase = flyteapp.Status_DEPLOYMENT_STATUS_FAILED
470-
if condition := ksvc.Status.GetCondition(servingv1.ServiceConditionReady); condition != nil {
471-
message = condition.Message
489+
if c := ksvc.Status.GetCondition(servingv1.ServiceConditionReady); c != nil {
490+
message = c.Message
472491
}
473492
case ksvc.Status.LatestCreatedRevisionName != ksvc.Status.LatestReadyRevisionName:
474493
phase = flyteapp.Status_DEPLOYMENT_STATUS_DEPLOYING

‎app/internal/k8s/app_client_test.go‎

Lines changed: 39 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -283,13 +283,51 @@ func TestList(t *testing.T) {
283283
},
284284
}
285285

286-
apps, err := c.List(context.Background(), "proj", "dev")
286+
apps, nextToken, err := c.List(context.Background(), "proj", "dev", "", 0, "")
287287
require.NoError(t, err)
288+
assert.Empty(t, nextToken)
288289
require.Len(t, apps, 1)
289290
assert.Equal(t, "proj", apps[0].Metadata.Id.Project)
290291
assert.Equal(t, "app1", apps[0].Metadata.Id.Name)
291292
}
292293

294+
func TestList_ByAppName(t *testing.T) {
295+
s := testScheme(t)
296+
ksvc1 := &servingv1.Service{
297+
ObjectMeta: metav1.ObjectMeta{
298+
Name: "app1",
299+
Namespace: "proj-dev",
300+
Labels: map[string]string{
301+
labelAppManaged: "true",
302+
labelProject: "proj",
303+
labelDomain: "dev",
304+
labelAppName: "app1",
305+
},
306+
Annotations: map[string]string{annotationAppID: "proj/dev/app1"},
307+
},
308+
}
309+
ksvc2 := &servingv1.Service{
310+
ObjectMeta: metav1.ObjectMeta{
311+
Name: "app2",
312+
Namespace: "proj-dev",
313+
Labels: map[string]string{
314+
labelAppManaged: "true",
315+
labelProject: "proj",
316+
labelDomain: "dev",
317+
labelAppName: "app2",
318+
},
319+
Annotations: map[string]string{annotationAppID: "proj/dev/app2"},
320+
},
321+
}
322+
fc := fake.NewClientBuilder().WithScheme(s).WithObjects(ksvc1, ksvc2).Build()
323+
c := &AppK8sClient{k8sClient: fc, cfg: &config.AppConfig{}}
324+
325+
apps, _, err := c.List(context.Background(), "proj", "dev", "app1", 0, "")
326+
require.NoError(t, err)
327+
require.Len(t, apps, 1)
328+
assert.Equal(t, "app1", apps[0].Metadata.Id.Name)
329+
}
330+
293331
func TestGetReplicas(t *testing.T) {
294332
s := testScheme(t)
295333
pod := &corev1.Pod{

0 commit comments

Comments
 (0)