diff --git a/workspaces/backend/api/app.go b/workspaces/backend/api/app.go index 274138aaa..133c90ec7 100644 --- a/workspaces/backend/api/app.go +++ b/workspaces/backend/api/app.go @@ -26,6 +26,7 @@ import ( "k8s.io/apimachinery/pkg/runtime/serializer" "k8s.io/apiserver/pkg/authentication/authenticator" "k8s.io/apiserver/pkg/authorization/authorizer" + "k8s.io/client-go/kubernetes" "sigs.k8s.io/controller-runtime/pkg/client" "github.com/kubeflow/notebooks/workspaces/backend/api/constants" @@ -54,6 +55,7 @@ func NewApp( scheme *runtime.Scheme, reqAuthN authenticator.Request, reqAuthZ authorizer.Authorizer, + clientset kubernetes.Interface, ) (*App, error) { // TODO: log the configuration on startup @@ -68,7 +70,7 @@ func NewApp( app := &App{ Config: cfg, logger: logger, - repositories: repositories.NewRepositories(cfg, cl, configMapClient), + repositories: repositories.NewRepositories(cfg, cl, configMapClient, clientset), Scheme: scheme, StrictYamlSerializer: yamlSerializerInfo.StrictSerializer, RequestAuthN: reqAuthN, @@ -106,6 +108,7 @@ func (a *App) Routes() http.Handler { router.DELETE(constants.WorkspacesByNamePath, a.DeleteWorkspaceHandler) router.POST(constants.PauseWorkspacePath, a.PauseActionWorkspaceHandler) router.GET(constants.WorkspacePodTemplateDetailsPath, a.GetWorkspacePodTemplateDetailsHandler) + router.GET(constants.WorkspacePodTemplatePodLogsBatchPath, a.GetWorkspacePodTemplateLogsHandler) // workspacekinds router.GET(constants.AllWorkspaceKindsPath, a.GetWorkspaceKindsHandler) diff --git a/workspaces/backend/api/constants/paths.go b/workspaces/backend/api/constants/paths.go index 5e681870a..276aac650 100644 --- a/workspaces/backend/api/constants/paths.go +++ b/workspaces/backend/api/constants/paths.go @@ -24,12 +24,13 @@ const ( HealthCheckPath = PathPrefix + "/healthcheck" // workspaces - AllWorkspacesPath = PathPrefix + "/workspaces" - WorkspacesByNamespacePath = AllWorkspacesPath + "/:" + NamespacePathParam - WorkspacesByNamePath = AllWorkspacesPath + "/:" + NamespacePathParam + "/:" + ResourceNamePathParam - WorkspaceActionsPath = WorkspacesByNamePath + "/actions" - PauseWorkspacePath = WorkspaceActionsPath + "/pause" - WorkspacePodTemplateDetailsPath = WorkspacesByNamePath + "/podtemplate/details" + AllWorkspacesPath = PathPrefix + "/workspaces" + WorkspacesByNamespacePath = AllWorkspacesPath + "/:" + NamespacePathParam + WorkspacesByNamePath = AllWorkspacesPath + "/:" + NamespacePathParam + "/:" + ResourceNamePathParam + WorkspaceActionsPath = WorkspacesByNamePath + "/actions" + PauseWorkspacePath = WorkspaceActionsPath + "/pause" + WorkspacePodTemplateDetailsPath = WorkspacesByNamePath + "/podtemplate/details" + WorkspacePodTemplatePodLogsBatchPath = WorkspacesByNamePath + "/podtemplate/logs/batch" // workspacekinds AllWorkspaceKindsPath = PathPrefix + "/workspacekinds" diff --git a/workspaces/backend/api/constants/query_params.go b/workspaces/backend/api/constants/query_params.go index 90da0ceee..053975adb 100644 --- a/workspaces/backend/api/constants/query_params.go +++ b/workspaces/backend/api/constants/query_params.go @@ -17,4 +17,8 @@ package constants const ( NamespaceQueryParam = "namespace" NamespaceFilterQueryParam = "namespaceFilter" + ContainerQueryParam = "container" + TailLinesQueryParam = "tailLines" + PreviousQueryParam = "previous" + SinceTimeQueryParam = "sinceTime" ) diff --git a/workspaces/backend/api/suite_test.go b/workspaces/backend/api/suite_test.go index 07f0b6775..813bd0d80 100644 --- a/workspaces/backend/api/suite_test.go +++ b/workspaces/backend/api/suite_test.go @@ -32,6 +32,7 @@ import ( rbacv1 "k8s.io/api/rbac/v1" "k8s.io/apimachinery/pkg/api/resource" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/kubernetes" "k8s.io/client-go/kubernetes/scheme" "k8s.io/client-go/rest" metricsv1beta1 "k8s.io/metrics/pkg/apis/metrics/v1beta1" @@ -148,9 +149,13 @@ var _ = BeforeSuite(func() { imageSourceConfigMapClient, err := helper.BuildImageSourceConfigMapClient(k8sManager) Expect(err).NotTo(HaveOccurred()) + By("creating the Kubernetes clientset") + clientset, err := kubernetes.NewForConfig(cfg) + Expect(err).NotTo(HaveOccurred()) + By("creating the application") // NOTE: we use the `k8sClient` rather than `k8sManager.GetClient()` to avoid race conditions with the cached client - a, err = NewApp(&config.EnvConfig{}, appLogger, k8sClient, imageSourceConfigMapClient, k8sManager.GetScheme(), reqAuthN, reqAuthZ) + a, err = NewApp(&config.EnvConfig{}, appLogger, k8sClient, imageSourceConfigMapClient, k8sManager.GetScheme(), reqAuthN, reqAuthZ, clientset) Expect(err).NotTo(HaveOccurred()) go func() { diff --git a/workspaces/backend/api/workspace_podtemplate_logs_handler.go b/workspaces/backend/api/workspace_podtemplate_logs_handler.go new file mode 100644 index 000000000..5125d5c70 --- /dev/null +++ b/workspaces/backend/api/workspace_podtemplate_logs_handler.go @@ -0,0 +1,158 @@ +/* +Copyright 2024. + +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. +*/ + +package api + +import ( + "errors" + "io" + "net/http" + + "github.com/julienschmidt/httprouter" + "k8s.io/apimachinery/pkg/util/validation/field" + + "github.com/kubeflow/notebooks/workspaces/backend/api/constants" + "github.com/kubeflow/notebooks/workspaces/backend/internal/auth" + "github.com/kubeflow/notebooks/workspaces/backend/internal/helper" + models "github.com/kubeflow/notebooks/workspaces/backend/internal/models/workspaces/podtemplate/logs" + repository "github.com/kubeflow/notebooks/workspaces/backend/internal/repositories/podlogs" +) + +// GetWorkspacePodTemplateLogsHandler returns a point-in-time snapshot of container logs for a workspace pod. +// +// @Summary Get workspace container logs (batch) +// @Description Returns a point-in-time snapshot of container logs for the workspace pod as a raw text/plain stream proxied directly from the Kubernetes pod logs API. +// @Tags workspaces +// @ID getWorkspacePodTemplateLogsBatch +// @Produce plain +// @Param namespace path string true "Namespace of the workspace" extensions(x-example=kubeflow-user-example-com) +// @Param name path string true "Name of the workspace" extensions(x-example=my-workspace) +// @Param container query string false "Target container name. Defaults to the primary (main) container." +// @Param tailLines query integer false "Number of lines from the end of the log to return. Defaults to 1000." +// @Param sinceTime query string false "Only return logs after this RFC3339 timestamp (e.g. 2026-07-15T10:30:00Z)." +// @Param previous query boolean false "If true, returns logs from the previous terminated container instance." +// @Success 200 {string} string "Raw container log stream (text/plain)." +// @Failure 400 {object} ErrorEnvelope "Bad Request. Container not found, pod not running, container not started, or no previous logs available." +// @Failure 401 {object} ErrorEnvelope "Unauthorized." +// @Failure 403 {object} ErrorEnvelope "Forbidden." +// @Failure 404 {object} ErrorEnvelope "Workspace not found." +// @Failure 422 {object} ErrorEnvelope "Unprocessable Entity. Validation error." +// @Failure 500 {object} ErrorEnvelope "Internal server error." +// @Router /workspaces/{namespace}/{name}/podtemplate/logs/batch [get] +func (a *App) GetWorkspacePodTemplateLogsHandler(w http.ResponseWriter, r *http.Request, ps httprouter.Params) { + namespace := ps.ByName(constants.NamespacePathParam) + workspaceName := ps.ByName(constants.ResourceNamePathParam) + + // validate path parameters + var valErrs field.ErrorList //nolint:prealloc + valErrs = append(valErrs, helper.ValidateKubernetesNamespaceName(field.NewPath(constants.NamespacePathParam), namespace)...) + valErrs = append(valErrs, helper.ValidateWorkspaceName(field.NewPath(constants.ResourceNamePathParam), workspaceName)...) + if len(valErrs) > 0 { + a.failedValidationResponse(w, r, errMsgPathParamsInvalid, valErrs, nil) + return + } + + // parse and validate query parameters + opts, valErrs := parseLogOptions(r) + if len(valErrs) > 0 { + a.failedValidationResponse(w, r, errMsgQueryParamsInvalid, valErrs, nil) + return + } + + // =========================== AUTH =========================== + authPolicies := []*auth.ResourcePolicy{ + auth.NewResourcePolicy(auth.VerbGet, auth.Workspaces, auth.ResourcePolicyResourceMeta{Namespace: namespace, Name: workspaceName}), + } + if _, ok := a.requireAuth(w, r, authPolicies); !ok { + return + } + // ============================================================ + + stream, err := a.repositories.PodLogs.OpenLogStream(r.Context(), namespace, workspaceName, opts) + if err != nil { + switch { + case errors.Is(err, repository.ErrWorkspaceNotFound): + a.notFoundResponse(w, r) + case errors.Is(err, repository.ErrPreviousLogsNotFound): + a.badRequestResponse(w, r, err) + case errors.Is(err, repository.ErrPodNotRunning): + a.badRequestResponse(w, r, err) + case errors.Is(err, repository.ErrContainerNotFound): + a.badRequestResponse(w, r, err) + case errors.Is(err, repository.ErrContainerNotRunning): + a.badRequestResponse(w, r, err) + default: + a.serverErrorResponse(w, r, err) + } + return + } + defer func() { _ = stream.Close() }() + + // Success responses are always a raw text/plain stream. + w.Header().Set("Content-Type", "text/plain") + w.WriteHeader(http.StatusOK) + + // Proxy the log stream directly to the client. + if _, err := io.Copy(w, stream); err != nil { + a.logger.Error("error while streaming workspace logs", "error", err, "namespace", namespace, "workspace", workspaceName) + } +} + +// parseLogOptions parses and validates the log-related query parameters. +func parseLogOptions(r *http.Request) (*models.LogOptions, field.ErrorList) { + var valErrs field.ErrorList + query := r.URL.Query() + + opts := &models.LogOptions{} + + if raw := query.Get(constants.ContainerQueryParam); raw != "" { + errs := helper.ValidateKubernetesContainersName(field.NewPath(constants.ContainerQueryParam), raw) + if len(errs) > 0 { + valErrs = append(valErrs, errs...) + } else { + opts.Container = raw + } + } + + if raw := query.Get(constants.TailLinesQueryParam); raw != "" { + tail, errs := helper.ValidateFieldIsPositiveInt64(field.NewPath(constants.TailLinesQueryParam), raw) + if len(errs) > 0 { + valErrs = append(valErrs, errs...) + } else { + opts.TailLines = tail + } + } + + if raw := query.Get(constants.PreviousQueryParam); raw != "" { + previous, errs := helper.ValidateFieldIsBool(field.NewPath(constants.PreviousQueryParam), raw) + if len(errs) > 0 { + valErrs = append(valErrs, errs...) + } else { + opts.Previous = previous + } + } + + if raw := query.Get(constants.SinceTimeQueryParam); raw != "" { + sinceTime, errs := helper.ValidateFieldIsRFC3339Time(field.NewPath(constants.SinceTimeQueryParam), raw) + if len(errs) > 0 { + valErrs = append(valErrs, errs...) + } else { + opts.SinceTime = &sinceTime + } + } + + return opts, valErrs +} diff --git a/workspaces/backend/api/workspace_podtemplate_logs_handler_test.go b/workspaces/backend/api/workspace_podtemplate_logs_handler_test.go new file mode 100644 index 000000000..c8b070192 --- /dev/null +++ b/workspaces/backend/api/workspace_podtemplate_logs_handler_test.go @@ -0,0 +1,256 @@ +/* +Copyright 2024. + +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. +*/ + +package api + +import ( + "encoding/json" + "fmt" + "io" + "net/http" + "net/http/httptest" + "strings" + + "github.com/julienschmidt/httprouter" + kubefloworgv1beta1 "github.com/kubeflow/notebooks/workspaces/controller/api/v1beta1" + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + + "github.com/kubeflow/notebooks/workspaces/backend/api/constants" + repository "github.com/kubeflow/notebooks/workspaces/backend/internal/repositories/podlogs" +) + +var _ = Describe("Workspace Logs Handler", func() { + + // buildLogsRequest constructs the HTTP request and httprouter params for the + // batch logs endpoint, applying the given raw query string (may be empty). + buildLogsRequest := func(namespace, workspaceName, rawQuery string) (*http.Request, httprouter.Params) { + path := strings.Replace(constants.WorkspacePodTemplatePodLogsBatchPath, ":"+constants.NamespacePathParam, namespace, 1) + path = strings.Replace(path, ":"+constants.ResourceNamePathParam, workspaceName, 1) + if rawQuery != "" { + path += "?" + rawQuery + } + + req, err := http.NewRequest(http.MethodGet, path, http.NoBody) + Expect(err).NotTo(HaveOccurred()) + req.Header.Set(userIdHeader, adminUser) + + ps := httprouter.Params{ + httprouter.Param{Key: constants.NamespacePathParam, Value: namespace}, + httprouter.Param{Key: constants.ResourceNamePathParam, Value: workspaceName}, + } + return req, ps + } + + Context("with invalid query parameters", func() { + + It("should return 422 when tailLines is not a positive integer", func() { + By("creating the HTTP request with a non-integer tailLines") + req, ps := buildLogsRequest("logs-ns", "workspace-logs", "tailLines=abc") + + By("executing GetWorkspacePodTemplateLogsHandler") + rr := httptest.NewRecorder() + a.GetWorkspacePodTemplateLogsHandler(rr, req, ps) + rs := rr.Result() + defer rs.Body.Close() + + By("verifying status is 422 Unprocessable Entity") + Expect(rs.StatusCode).To(Equal(http.StatusUnprocessableEntity)) + + By("verifying the error message indicates a query parameter validation failure") + var response ErrorEnvelope + Expect(json.Unmarshal(rr.Body.Bytes(), &response)).To(Succeed()) + Expect(response.Error.Message).To(Equal(errMsgQueryParamsInvalid)) + Expect(response.Error.Cause.ValidationErrors).NotTo(BeEmpty()) + Expect(response.Error.Cause.ValidationErrors[0].Field).To(Equal(constants.TailLinesQueryParam)) + }) + + It("should return 422 when tailLines is zero or negative", func() { + By("creating the HTTP request with a non-positive tailLines") + req, ps := buildLogsRequest("logs-ns", "workspace-logs", "tailLines=0") + + By("executing GetWorkspacePodTemplateLogsHandler") + rr := httptest.NewRecorder() + a.GetWorkspacePodTemplateLogsHandler(rr, req, ps) + rs := rr.Result() + defer rs.Body.Close() + + By("verifying status is 422 Unprocessable Entity") + Expect(rs.StatusCode).To(Equal(http.StatusUnprocessableEntity)) + + By("verifying the error message indicates a query parameter validation failure") + var response ErrorEnvelope + Expect(json.Unmarshal(rr.Body.Bytes(), &response)).To(Succeed()) + Expect(response.Error.Message).To(Equal(errMsgQueryParamsInvalid)) + Expect(response.Error.Cause.ValidationErrors).NotTo(BeEmpty()) + Expect(response.Error.Cause.ValidationErrors[0].Field).To(Equal(constants.TailLinesQueryParam)) + }) + + It("should return 422 when previous is not a boolean", func() { + By("creating the HTTP request with a non-boolean previous") + req, ps := buildLogsRequest("logs-ns", "workspace-logs", "previous=maybe") + + By("executing GetWorkspacePodTemplateLogsHandler") + rr := httptest.NewRecorder() + a.GetWorkspacePodTemplateLogsHandler(rr, req, ps) + rs := rr.Result() + defer rs.Body.Close() + + By("verifying status is 422 Unprocessable Entity") + Expect(rs.StatusCode).To(Equal(http.StatusUnprocessableEntity)) + + By("verifying the error message indicates a query parameter validation failure") + var response ErrorEnvelope + Expect(json.Unmarshal(rr.Body.Bytes(), &response)).To(Succeed()) + Expect(response.Error.Message).To(Equal(errMsgQueryParamsInvalid)) + Expect(response.Error.Cause.ValidationErrors).NotTo(BeEmpty()) + Expect(response.Error.Cause.ValidationErrors[0].Field).To(Equal(constants.PreviousQueryParam)) + }) + + It("should return 422 when sinceTime is not a valid RFC3339 timestamp", func() { + By("creating the HTTP request with an invalid sinceTime") + req, ps := buildLogsRequest("logs-ns", "workspace-logs", "sinceTime=not-a-timestamp") + + By("executing GetWorkspacePodTemplateLogsHandler") + rr := httptest.NewRecorder() + a.GetWorkspacePodTemplateLogsHandler(rr, req, ps) + rs := rr.Result() + defer rs.Body.Close() + + By("verifying status is 422 Unprocessable Entity") + Expect(rs.StatusCode).To(Equal(http.StatusUnprocessableEntity)) + + By("verifying the error message indicates a query parameter validation failure") + var response ErrorEnvelope + Expect(json.Unmarshal(rr.Body.Bytes(), &response)).To(Succeed()) + Expect(response.Error.Message).To(Equal(errMsgQueryParamsInvalid)) + Expect(response.Error.Cause.ValidationErrors).NotTo(BeEmpty()) + Expect(response.Error.Cause.ValidationErrors[0].Field).To(Equal(constants.SinceTimeQueryParam)) + }) + + It("should return 422 when container is not a valid DNS1123 label", func() { + By("creating the HTTP request with an invalid container name") + req, ps := buildLogsRequest("logs-ns", "workspace-logs", "container=BAD_NAME!") + + By("executing GetWorkspacePodTemplateLogsHandler") + rr := httptest.NewRecorder() + a.GetWorkspacePodTemplateLogsHandler(rr, req, ps) + rs := rr.Result() + defer rs.Body.Close() + + By("verifying status is 422 Unprocessable Entity") + Expect(rs.StatusCode).To(Equal(http.StatusUnprocessableEntity)) + + By("verifying the error message indicates a query parameter validation failure") + var response ErrorEnvelope + Expect(json.Unmarshal(rr.Body.Bytes(), &response)).To(Succeed()) + Expect(response.Error.Message).To(Equal(errMsgQueryParamsInvalid)) + Expect(response.Error.Cause.ValidationErrors).NotTo(BeEmpty()) + Expect(response.Error.Cause.ValidationErrors[0].Field).To(Equal(constants.ContainerQueryParam)) + }) + }) + + Context("with a non-existent workspace", func() { + + It("should return 404 with a descriptive message when the workspace does not exist", func() { + By("creating the HTTP request for a missing workspace") + req, ps := buildLogsRequest("logs-ns", "does-not-exist", "") + + By("executing GetWorkspacePodTemplateLogsHandler") + rr := httptest.NewRecorder() + a.GetWorkspacePodTemplateLogsHandler(rr, req, ps) + rs := rr.Result() + defer rs.Body.Close() + + By("verifying status is 404 Not Found") + Expect(rs.StatusCode).To(Equal(http.StatusNotFound)) + + By("verifying the response carries the generic 'not found' message") + body, err := io.ReadAll(rs.Body) + Expect(err).NotTo(HaveOccurred()) + var envelope ErrorEnvelope + Expect(json.Unmarshal(body, &envelope)).To(Succeed()) + Expect(envelope.Error).NotTo(BeNil()) + Expect(envelope.Error.Message).To(Equal(errMsgNotFound)) + }) + }) + + Context("with an existing Workspace that has no running pod", Serial, Ordered, func() { + const namespaceName = "logs-nopod-ns" + var ( + workspaceName string + workspaceKindName string + ) + + BeforeAll(func() { + uniqueName := "logs-nopod-test" + workspaceName = fmt.Sprintf("workspace-%s", uniqueName) + workspaceKindName = fmt.Sprintf("workspacekind-%s", uniqueName) + + By("creating the Namespace") + Expect(k8sClient.Create(ctx, &corev1.Namespace{ + ObjectMeta: metav1.ObjectMeta{Name: namespaceName}, + })).To(Succeed()) + + By("creating the WorkspaceKind") + Expect(k8sClient.Create(ctx, NewExampleWorkspaceKind(workspaceKindName))).To(Succeed()) + + By("creating the Workspace (status.podTemplatePod.name is empty)") + Expect(k8sClient.Create(ctx, NewExampleWorkspace(workspaceName, namespaceName, workspaceKindName))).To(Succeed()) + }) + + AfterAll(func() { + By("deleting the Workspace") + Expect(k8sClient.Delete(ctx, &kubefloworgv1beta1.Workspace{ + ObjectMeta: metav1.ObjectMeta{Name: workspaceName, Namespace: namespaceName}, + })).To(Succeed()) + + By("deleting the WorkspaceKind") + Expect(k8sClient.Delete(ctx, &kubefloworgv1beta1.WorkspaceKind{ + ObjectMeta: metav1.ObjectMeta{Name: workspaceKindName}, + })).To(Succeed()) + + By("deleting the Namespace") + Expect(k8sClient.Delete(ctx, &corev1.Namespace{ + ObjectMeta: metav1.ObjectMeta{Name: namespaceName}, + })).To(Succeed()) + }) + + It("should return 400 when the workspace pod is not running", func() { + By("creating the HTTP request") + req, ps := buildLogsRequest(namespaceName, workspaceName, "") + + By("executing GetWorkspacePodTemplateLogsHandler") + rr := httptest.NewRecorder() + a.GetWorkspacePodTemplateLogsHandler(rr, req, ps) + rs := rr.Result() + defer rs.Body.Close() + + By("verifying status is 400 Bad Request") + Expect(rs.StatusCode).To(Equal(http.StatusBadRequest)) + + By("verifying the response carries the specific 'workspace pod is not running' message") + body, err := io.ReadAll(rs.Body) + Expect(err).NotTo(HaveOccurred()) + var envelope ErrorEnvelope + Expect(json.Unmarshal(body, &envelope)).To(Succeed()) + Expect(envelope.Error).NotTo(BeNil()) + Expect(envelope.Error.Message).To(Equal(repository.ErrPodNotRunning.Error())) + }) + }) +}) diff --git a/workspaces/backend/cmd/main.go b/workspaces/backend/cmd/main.go index 6167b9927..8c17e116d 100644 --- a/workspaces/backend/cmd/main.go +++ b/workspaces/backend/cmd/main.go @@ -22,6 +22,7 @@ import ( "os" "strconv" + "k8s.io/client-go/kubernetes" ctrl "sigs.k8s.io/controller-runtime" application "github.com/kubeflow/notebooks/workspaces/backend/api" @@ -150,6 +151,12 @@ func main() { os.Exit(1) } + clientset, err := kubernetes.NewForConfig(kubeconfig) + if err != nil { + logger.Error("failed to create Kubernetes clientset", "error", err) + os.Exit(1) + } + // Create the request authenticator reqAuthN, err := auth.NewRequestAuthenticator(cfg.UserIdHeader, cfg.UserIdPrefix, cfg.GroupsHeader) if err != nil { @@ -181,6 +188,7 @@ func main() { mgr.GetScheme(), reqAuthN, reqAuthZ, + clientset, ) if err != nil { logger.Error("failed to create app", "error", err) diff --git a/workspaces/backend/internal/helper/validation.go b/workspaces/backend/internal/helper/validation.go index 1ad77c164..956c82ca8 100644 --- a/workspaces/backend/internal/helper/validation.go +++ b/workspaces/backend/internal/helper/validation.go @@ -21,7 +21,9 @@ import ( "encoding/base64" "errors" "fmt" + "strconv" "strings" + "time" kubefloworgv1beta1 "github.com/kubeflow/notebooks/workspaces/controller/api/v1beta1" corev1 "k8s.io/api/core/v1" @@ -257,7 +259,7 @@ func ValidateKubernetesStorageClassIsUsable(ctx context.Context, k8sClient clien // ValidateFieldIsDNS1123Label validates a field contains an RCF 1123 DNS label. // USED FOR: -// - names of: Namespaces, Services, etc. +// - names of: Namespaces, Services, Containers, etc. func ValidateFieldIsDNS1123Label(path *field.Path, value string) field.ErrorList { var errs field.ErrorList @@ -283,6 +285,11 @@ func ValidateKubernetesServicesName(path *field.Path, value string) field.ErrorL return ValidateFieldIsDNS1123Label(path, value) } +// ValidateKubernetesContainersName validates a field contains a valid Kubernetes container name. +func ValidateKubernetesContainersName(path *field.Path, value string) field.ErrorList { + return ValidateFieldIsDNS1123Label(path, value) +} + // ValidateKubernetesAnnotations validates a map of Kubernetes annotations. func ValidateKubernetesAnnotations(path *field.Path, annotations map[string]string) field.ErrorList { return apivalidation.ValidateAnnotations(annotations, path) @@ -320,3 +327,46 @@ func ValidateFieldIsSecretBase64Value(path *field.Path, value string) field.Erro return errs } + +// ValidateFieldIsPositiveInt64 parses value as a base-10 int64 and validates it +// is a positive integer (> 0). On success it returns the parsed value and a nil +// error list. +func ValidateFieldIsPositiveInt64(path *field.Path, value string) (int64, field.ErrorList) { + var errs field.ErrorList + + parsed, err := strconv.ParseInt(value, 10, 64) + if err != nil || parsed <= 0 { + errs = append(errs, field.Invalid(path, value, "must be a positive integer")) + return 0, errs + } + + return parsed, errs +} + +// ValidateFieldIsBool parses value as a boolean. On success it returns the parsed +// value and a nil error list. +func ValidateFieldIsBool(path *field.Path, value string) (bool, field.ErrorList) { + var errs field.ErrorList + + parsed, err := strconv.ParseBool(value) + if err != nil { + errs = append(errs, field.Invalid(path, value, "must be a boolean")) + return false, errs + } + + return parsed, errs +} + +// ValidateFieldIsRFC3339Time parses value as an RFC3339 timestamp. On success it +// returns the parsed value and a nil error list. +func ValidateFieldIsRFC3339Time(path *field.Path, value string) (metav1.Time, field.ErrorList) { + var errs field.ErrorList + + parsed, err := time.Parse(time.RFC3339, value) + if err != nil { + errs = append(errs, field.Invalid(path, value, "must be a valid RFC3339 timestamp")) + return metav1.Time{}, errs + } + + return metav1.NewTime(parsed), errs +} diff --git a/workspaces/backend/internal/helper/validation_test.go b/workspaces/backend/internal/helper/validation_test.go index 7f370f10b..152ffae36 100644 --- a/workspaces/backend/internal/helper/validation_test.go +++ b/workspaces/backend/internal/helper/validation_test.go @@ -19,6 +19,7 @@ package helper import ( "net/http" "testing" + "time" kubefloworgv1beta1 "github.com/kubeflow/notebooks/workspaces/controller/api/v1beta1" . "github.com/onsi/ginkgo/v2" @@ -853,4 +854,174 @@ var _ = Describe("Validation Helper Functions", func() { }) } }) + + Describe("ValidateFieldIsPositiveInt64", func() { + const fieldName = "tailLines" + + type testCase struct { + description string + value string + expectValue int64 + expectError bool + } + + testCases := []testCase{ + { + description: "should parse a positive integer", + value: "15", + expectValue: 15, + expectError: false, + }, + { + description: "should return error for zero", + value: "0", + expectError: true, + }, + { + description: "should return error for a negative integer", + value: "-5", + expectError: true, + }, + { + description: "should return error for a non-integer", + value: "abc", + expectError: true, + }, + { + description: "should return error for an empty string", + value: "", + expectError: true, + }, + } + + for _, tc := range testCases { + It(tc.description, func() { + path := field.NewPath(fieldName) + value, errs := ValidateFieldIsPositiveInt64(path, tc.value) + if tc.expectError { + Expect(errs).To(HaveLen(1)) + Expect(errs[0].Type).To(Equal(field.ErrorTypeInvalid)) + Expect(errs[0].Field).To(Equal(fieldName)) + Expect(value).To(BeZero()) + } else { + Expect(errs).To(BeEmpty()) + Expect(value).To(Equal(tc.expectValue)) + } + }) + } + }) + + Describe("ValidateFieldIsBool", func() { + const fieldName = "previous" + + type testCase struct { + description string + value string + expectValue bool + expectError bool + } + + testCases := []testCase{ + { + description: "should parse true", + value: "true", + expectValue: true, + expectError: false, + }, + { + description: "should parse false", + value: "false", + expectValue: false, + expectError: false, + }, + { + description: "should parse the shorthand 1 as true", + value: "1", + expectValue: true, + expectError: false, + }, + { + description: "should return error for a non-boolean", + value: "maybe", + expectError: true, + }, + { + description: "should return error for an empty string", + value: "", + expectError: true, + }, + } + + for _, tc := range testCases { + It(tc.description, func() { + path := field.NewPath(fieldName) + value, errs := ValidateFieldIsBool(path, tc.value) + if tc.expectError { + Expect(errs).To(HaveLen(1)) + Expect(errs[0].Type).To(Equal(field.ErrorTypeInvalid)) + Expect(errs[0].Field).To(Equal(fieldName)) + Expect(value).To(BeFalse()) + } else { + Expect(errs).To(BeEmpty()) + Expect(value).To(Equal(tc.expectValue)) + } + }) + } + }) + + Describe("ValidateFieldIsRFC3339Time", func() { + const fieldName = "sinceTime" + + type testCase struct { + description string + value string + expectError bool + } + + testCases := []testCase{ + { + description: "should parse a valid RFC3339 timestamp", + value: "2026-07-15T10:30:00Z", + expectError: false, + }, + { + description: "should parse a valid RFC3339 timestamp with offset", + value: "2026-07-15T10:30:00+02:00", + expectError: false, + }, + { + description: "should return error for a non-timestamp", + value: "not-a-timestamp", + expectError: true, + }, + { + description: "should return error for a date-only string (not RFC3339)", + value: "2026-07-15", + expectError: true, + }, + { + description: "should return error for an empty string", + value: "", + expectError: true, + }, + } + + for _, tc := range testCases { + It(tc.description, func() { + path := field.NewPath(fieldName) + value, errs := ValidateFieldIsRFC3339Time(path, tc.value) + if tc.expectError { + Expect(errs).To(HaveLen(1)) + Expect(errs[0].Type).To(Equal(field.ErrorTypeInvalid)) + Expect(errs[0].Field).To(Equal(fieldName)) + Expect(value.IsZero()).To(BeTrue()) + } else { + Expect(errs).To(BeEmpty()) + expected, perr := time.Parse(time.RFC3339, tc.value) + Expect(perr).NotTo(HaveOccurred()) + Expect(value.Time).To(BeTemporally("==", expected)) + } + }) + } + }) }) diff --git a/workspaces/backend/internal/models/workspaces/podtemplate/logs/types.go b/workspaces/backend/internal/models/workspaces/podtemplate/logs/types.go new file mode 100644 index 000000000..b6e46c18a --- /dev/null +++ b/workspaces/backend/internal/models/workspaces/podtemplate/logs/types.go @@ -0,0 +1,34 @@ +/* +Copyright 2024. + +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. +*/ + +package logs + +import ( + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" +) + +type LogOptions struct { + // The name of the container to retrieve logs from. + // If omitted, defaults to the workspace's primary container (named "main"). + Container string + // The number of lines to retrieve from the end of the logs. + // By default, the value is 1000. + TailLines int64 + // If true, returns logs from the previous terminated container instance. + Previous bool + // If specified, returns logs since the given time. + SinceTime *metav1.Time +} diff --git a/workspaces/backend/internal/repositories/podlogs/repo.go b/workspaces/backend/internal/repositories/podlogs/repo.go new file mode 100644 index 000000000..f1a4be908 --- /dev/null +++ b/workspaces/backend/internal/repositories/podlogs/repo.go @@ -0,0 +1,224 @@ +/* +Copyright 2024. + +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. +*/ + +package podlogs + +import ( + "context" + "fmt" + "io" + + kubefloworgv1beta1 "github.com/kubeflow/notebooks/workspaces/controller/api/v1beta1" + corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/kubernetes" + "sigs.k8s.io/controller-runtime/pkg/client" + + "github.com/kubeflow/notebooks/workspaces/backend/internal/config" + + models "github.com/kubeflow/notebooks/workspaces/backend/internal/models/workspaces/podtemplate/logs" +) + +var ( + ErrWorkspaceNotFound = fmt.Errorf("workspace not found") + ErrPodNotRunning = fmt.Errorf("workspace pod is not running") + ErrContainerNotFound = fmt.Errorf("container not found in workspace pod") + ErrContainerNotRunning = fmt.Errorf("container has not started yet") + ErrPreviousLogsNotFound = fmt.Errorf("no logs found for the previous container instance") +) + +// Format-string templates for wrapped errors. +const ( + fmtOpenLogStreamFailed = "failed to open log stream for pod %s, container %s: %w" + fmtGetPodFailed = "failed to get pod %s: %w" +) + +var ( + // safeLimitBytes is the maximum number of bytes the Kubernetes API will return + // for a single log request, bounding the size of the proxied stream. + safeLimitBytes = int64(100 * 1024 * 1024) // 100 MB + // defaultTailLines is the default number of lines to retrieve from the end of the logs. + defaultTailLines = int64(1000) +) + +// primaryContainerName is the name the controller hard-codes for a workspace's +// primary container. When no container is requested we default to it so +// that logs never accidentally target an injected sidecar (e.g. istio-proxy). +// +// This mirrors the controller's `workspacePodTemplateContainerName` constant defined in +// workspaces/controller/internal/controller/workspace_controller.go +const primaryContainerName = "main" + +type PodLogsRepository struct { + cfg *config.EnvConfig + client client.Client + clientset kubernetes.Interface +} + +func NewPodLogsRepository(cfg *config.EnvConfig, cl client.Client, clientset kubernetes.Interface) *PodLogsRepository { + return &PodLogsRepository{ + cfg: cfg, + client: cl, + clientset: clientset, + } +} + +func (r *PodLogsRepository) OpenLogStream(ctx context.Context, namespace, workspaceName string, opts *models.LogOptions) (io.ReadCloser, error) { + podName, containerName, err := r.resolvePodAndContainer(ctx, namespace, workspaceName, opts) + if err != nil { + return nil, err + } + + tailLines := opts.TailLines + if tailLines <= 0 { + tailLines = defaultTailLines + } + + req := r.clientset.CoreV1().Pods(namespace).GetLogs(podName, &corev1.PodLogOptions{ + Container: containerName, + TailLines: &tailLines, + LimitBytes: &safeLimitBytes, + Previous: opts.Previous, + Follow: false, + Timestamps: true, + SinceTime: opts.SinceTime, + }) + + stream, err := req.Stream(ctx) + if err != nil { + return nil, fmt.Errorf(fmtOpenLogStreamFailed, podName, containerName, err) + } + return stream, nil +} + +// containerExists reports whether a container with the given name exists among the +// pod's regular or init containers. Both are valid log sources (e.g. an istio-proxy +// native sidecar is an init container). +func containerExists(name string, containers, initContainers []kubefloworgv1beta1.WorkspacePodContainer) bool { + for _, c := range containers { + if c.Name == name { + return true + } + } + for _, c := range initContainers { + if c.Name == name { + return true + } + } + return false +} + +func (r *PodLogsRepository) resolvePodAndContainer(ctx context.Context, namespace, workspaceName string, opts *models.LogOptions) (string, string, error) { + workspace := &kubefloworgv1beta1.Workspace{} + if err := r.client.Get(ctx, client.ObjectKey{Namespace: namespace, Name: workspaceName}, workspace); err != nil { + if apierrors.IsNotFound(err) { + return "", "", ErrWorkspaceNotFound + } + return "", "", err + } + + podStatus := workspace.Status.PodTemplatePod + podName := podStatus.Name + if podName == "" { + return "", "", ErrPodNotRunning + } + + // Resolve the target container name. + containerName := opts.Container + if containerName != "" { + // A requested container must exist among the pod's containers. + if !containerExists(containerName, podStatus.Containers, podStatus.InitContainers) { + return "", "", ErrContainerNotFound + } + } else { + if len(podStatus.Containers) == 0 { + return "", "", ErrContainerNotRunning + } + if !containerExists(primaryContainerName, podStatus.Containers, nil) { + return "", "", ErrContainerNotFound + } + // None requested: default to the primary container, which the controller + // hard-codes as "main". + containerName = primaryContainerName + } + + // Inspect the live Pod status to make a semantic decision before opening the + // stream, so we can return a precise error instead of an opaque one from the + // Kubernetes log API. + if err := r.ensureLogsAvailable(ctx, namespace, podName, containerName, opts.Previous); err != nil { + return "", "", err + } + + return podName, containerName, nil +} + +// ensureLogsAvailable checks the live Pod status for the target container and +// returns a semantic error when logs cannot be served: +// - when previous is false and the container is still Waiting (never started), +// it returns ErrContainerNotRunning. +// - when previous is true and the container has no recorded previous terminated +// instance (LastTerminationState.Terminated is nil), it returns +// ErrPreviousLogsNotFound. +// +// The checker utilizes the clientset to fetch the live Pod status, +// which is more accurate than the Workspace's cached status. +func (r *PodLogsRepository) ensureLogsAvailable(ctx context.Context, namespace, podName, containerName string, previous bool) error { + pod, err := r.clientset.CoreV1().Pods(namespace).Get(ctx, podName, metav1.GetOptions{}) + if err != nil { + if apierrors.IsNotFound(err) { + // The Workspace status references a pod that no longer exists. + return ErrPodNotRunning + } + return fmt.Errorf(fmtGetPodFailed, podName, err) + } + + // checkStatus evaluates a single container status for the target container. + checkStatus := func(cs corev1.ContainerStatus) error { + if previous { + // A container with no recorded previous terminated instance has no + // previous logs to serve. + if cs.LastTerminationState.Terminated == nil { + return ErrPreviousLogsNotFound + } + return nil + } + // A container that is still Waiting has never started and has no logs yet. + if cs.State.Waiting != nil { + return ErrContainerNotRunning + } + return nil + } + + // Search the regular container statuses for the target container. + for _, cs := range pod.Status.ContainerStatuses { + if cs.Name != containerName { + continue + } + return checkStatus(cs) + } + + // Search the init container statuses for the target container (e.g. an + // istio-proxy native sidecar is an init container). + for _, cs := range pod.Status.InitContainerStatuses { + if cs.Name != containerName { + continue + } + return checkStatus(cs) + } + + return ErrContainerNotRunning +} diff --git a/workspaces/backend/internal/repositories/podlogs/repo_test.go b/workspaces/backend/internal/repositories/podlogs/repo_test.go new file mode 100644 index 000000000..a6ba7ab68 --- /dev/null +++ b/workspaces/backend/internal/repositories/podlogs/repo_test.go @@ -0,0 +1,403 @@ +/* +Copyright 2024. + +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. +*/ + +package podlogs + +import ( + "context" + "errors" + "io" + "testing" + + kubefloworgv1beta1 "github.com/kubeflow/notebooks/workspaces/controller/api/v1beta1" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + k8sfake "k8s.io/client-go/kubernetes/fake" + ctrlfake "sigs.k8s.io/controller-runtime/pkg/client/fake" + + "github.com/kubeflow/notebooks/workspaces/backend/internal/config" + "github.com/kubeflow/notebooks/workspaces/backend/internal/helper" + models "github.com/kubeflow/notebooks/workspaces/backend/internal/models/workspaces/podtemplate/logs" +) + +const ( + testNamespace = "test-ns" + testWorkspace = "test-ws" + testPodName = "ws-test-ws-0" +) + +// newWorkspace builds a Workspace CR with the given pod status containers. +func newWorkspace(podName string, containers ...string) *kubefloworgv1beta1.Workspace { + ws := &kubefloworgv1beta1.Workspace{ + ObjectMeta: metav1.ObjectMeta{Name: testWorkspace, Namespace: testNamespace}, + } + ws.Status.PodTemplatePod.Name = podName + for _, c := range containers { + ws.Status.PodTemplatePod.Containers = append( + ws.Status.PodTemplatePod.Containers, + kubefloworgv1beta1.WorkspacePodContainer{Name: c}, + ) + } + return ws +} + +// withInitContainers adds init containers (e.g. an istio-proxy native sidecar) to +// the workspace pod status. +func withInitContainers(ws *kubefloworgv1beta1.Workspace, initContainers ...string) *kubefloworgv1beta1.Workspace { + for _, c := range initContainers { + ws.Status.PodTemplatePod.InitContainers = append( + ws.Status.PodTemplatePod.InitContainers, + kubefloworgv1beta1.WorkspacePodContainer{Name: c}, + ) + } + return ws +} + +// runningContainerStatus returns a ContainerStatus for a container that has started. +func runningContainerStatus(name string) corev1.ContainerStatus { + return corev1.ContainerStatus{ + Name: name, + State: corev1.ContainerState{ + Running: &corev1.ContainerStateRunning{}, + }, + } +} + +// waitingContainerStatus returns a ContainerStatus for a container that is still Waiting. +func waitingContainerStatus(name, reason string) corev1.ContainerStatus { + return corev1.ContainerStatus{ + Name: name, + State: corev1.ContainerState{ + Waiting: &corev1.ContainerStateWaiting{Reason: reason}, + }, + } +} + +// waitingContainerStatusWithPrevious returns a ContainerStatus for a container +// that is currently Waiting but has a previous terminated instance, so requests +// for previous logs are considered available. +func waitingContainerStatusWithPrevious(name, reason string) corev1.ContainerStatus { + cs := waitingContainerStatus(name, reason) + cs.LastTerminationState = corev1.ContainerState{ + Terminated: &corev1.ContainerStateTerminated{ExitCode: 255}, + } + return cs +} + +// newPod builds a corev1.Pod (named testPodName) with the given regular and init +// container statuses. +func newPod(containerStatuses, initContainerStatuses []corev1.ContainerStatus) *corev1.Pod { + return &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{Name: testPodName, Namespace: testNamespace}, + Status: corev1.PodStatus{ + ContainerStatuses: containerStatuses, + InitContainerStatuses: initContainerStatuses, + }, + } +} + +// podFromWorkspace builds a corev1.Pod whose container statuses mirror the +// workspace's pod status, with every container reported as Running. This is the +// common case where all containers have started. +func podFromWorkspace(ws *kubefloworgv1beta1.Workspace) *corev1.Pod { + if ws == nil || ws.Status.PodTemplatePod.Name == "" { + return nil + } + var containers, initContainers []corev1.ContainerStatus + for _, c := range ws.Status.PodTemplatePod.Containers { + containers = append(containers, runningContainerStatus(c.Name)) + } + for _, c := range ws.Status.PodTemplatePod.InitContainers { + initContainers = append(initContainers, runningContainerStatus(c.Name)) + } + return newPod(containers, initContainers) +} + +// newRepo builds a PodLogsRepository backed by a fake controller-runtime client +// (seeded with the given workspace) and a fake Kubernetes clientset seeded with the +// given Pod (or nil for none). +func newRepo(t *testing.T, ws *kubefloworgv1beta1.Workspace, pod *corev1.Pod) *PodLogsRepository { + t.Helper() + + scheme, err := helper.BuildScheme() + if err != nil { + t.Fatalf("failed to build scheme: %v", err) + } + + builder := ctrlfake.NewClientBuilder().WithScheme(scheme) + if ws != nil { + builder = builder.WithObjects(ws) + } + cl := builder.Build() + + var clientset *k8sfake.Clientset + if pod != nil { + clientset = k8sfake.NewSimpleClientset(pod) + } else { + clientset = k8sfake.NewSimpleClientset() + } + + return NewPodLogsRepository(&config.EnvConfig{}, cl, clientset) +} + +func TestOpenLogStream(t *testing.T) { + testCases := []struct { + name string + ws *kubefloworgv1beta1.Workspace + // pod is the live Pod seeded into the fake clientset. Leave nil to seed no + // Pod (e.g. to test a missing pod); pass podFromWorkspace(ws) for the common + // all-Running case. + pod *corev1.Pod + opts *models.LogOptions + wantErr error // nil means success is expected + wantLogs bool // when success is expected, whether log content should be returned + }{ + { + name: "success with default container", + ws: newWorkspace(testPodName, "main", "istio-proxy"), + pod: podFromWorkspace(newWorkspace(testPodName, "main", "istio-proxy")), + opts: &models.LogOptions{}, + wantLogs: true, + }, + { + name: "success with specific container", + ws: newWorkspace(testPodName, "main", "istio-proxy"), + pod: podFromWorkspace(newWorkspace(testPodName, "main", "istio-proxy")), + opts: &models.LogOptions{Container: "istio-proxy"}, + wantLogs: true, + }, + { + name: "success with init container (native sidecar)", + ws: withInitContainers(newWorkspace(testPodName, "main"), "istio-proxy"), + pod: podFromWorkspace(withInitContainers(newWorkspace(testPodName, "main"), "istio-proxy")), + opts: &models.LogOptions{Container: "istio-proxy"}, + wantLogs: true, + }, + { + name: "workspace not found", + ws: nil, + opts: &models.LogOptions{}, + wantErr: ErrWorkspaceNotFound, + }, + { + name: "pod not running", + ws: newWorkspace("", "main"), + opts: &models.LogOptions{}, + wantErr: ErrPodNotRunning, + }, + { + name: "container not found", + ws: newWorkspace(testPodName, "main"), + opts: &models.LogOptions{Container: "does-not-exist"}, + wantErr: ErrContainerNotFound, + }, + { + name: "container not running when pod has no containers yet", + ws: newWorkspace(testPodName), // pod name set, but no containers listed + opts: &models.LogOptions{}, + wantErr: ErrContainerNotRunning, + }, + { + name: "container waiting returns not running (default container)", + ws: newWorkspace(testPodName, "main"), + pod: newPod([]corev1.ContainerStatus{ + waitingContainerStatus("main", "PodInitializing"), + }, nil), + opts: &models.LogOptions{}, + wantErr: ErrContainerNotRunning, + }, + { + name: "requested container waiting returns not running", + ws: newWorkspace(testPodName, "main", "istio-proxy"), + pod: newPod([]corev1.ContainerStatus{ + runningContainerStatus("main"), + waitingContainerStatus("istio-proxy", "ContainerCreating"), + }, nil), + opts: &models.LogOptions{Container: "istio-proxy"}, + wantErr: ErrContainerNotRunning, + }, + { + name: "waiting init container returns not running", + ws: withInitContainers(newWorkspace(testPodName, "main"), "istio-proxy"), + pod: newPod( + []corev1.ContainerStatus{runningContainerStatus("main")}, + []corev1.ContainerStatus{waitingContainerStatus("istio-proxy", "PodInitializing")}, + ), + opts: &models.LogOptions{Container: "istio-proxy"}, + wantErr: ErrContainerNotRunning, + }, + { + name: "no container status reported yet returns not running", + ws: newWorkspace(testPodName, "main"), + pod: newPod(nil, nil), + opts: &models.LogOptions{}, + wantErr: ErrContainerNotRunning, + }, + { + // pod left nil: the Workspace references a pod the clientset cannot find. + name: "live pod missing returns pod not running", + ws: newWorkspace(testPodName, "main"), + opts: &models.LogOptions{}, + wantErr: ErrPodNotRunning, + }, + { + name: "waiting current container is served when previous=true and a previous instance exists", + ws: newWorkspace(testPodName, "main"), + pod: newPod([]corev1.ContainerStatus{ + waitingContainerStatusWithPrevious("main", "CrashLoopBackOff"), + }, nil), + opts: &models.LogOptions{Previous: true}, + wantLogs: true, + }, + { + name: "previous=true with no previous terminated instance returns previous logs not found", + ws: newWorkspace(testPodName, "main"), + pod: newPod([]corev1.ContainerStatus{ + runningContainerStatus("main"), // running, but never restarted (no LastTerminationState) + }, nil), + opts: &models.LogOptions{Previous: true}, + wantErr: ErrPreviousLogsNotFound, + }, + { + name: "previous=true for an init container with no previous instance returns previous logs not found", + ws: withInitContainers(newWorkspace(testPodName, "main"), "istio-proxy"), + pod: newPod( + []corev1.ContainerStatus{runningContainerStatus("main")}, + []corev1.ContainerStatus{runningContainerStatus("istio-proxy")}, + ), + opts: &models.LogOptions{Container: "istio-proxy", Previous: true}, + wantErr: ErrPreviousLogsNotFound, + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + repo := newRepo(t, tc.ws, tc.pod) + + stream, err := repo.OpenLogStream(context.Background(), testNamespace, testWorkspace, tc.opts) + + if tc.wantErr != nil { + if !errors.Is(err, tc.wantErr) { + t.Fatalf("expected error %v, got: %v", tc.wantErr, err) + } + return + } + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + defer stream.Close() + + // The fake clientset returns a canned non-empty log body ("fake logs"), + // so we only assert that content was produced. + body, err := io.ReadAll(stream) + if err != nil { + t.Fatalf("unexpected error reading stream: %v", err) + } + if tc.wantLogs && len(body) == 0 { + t.Fatalf("expected at least some log content, got none") + } + }) + } +} + +func TestResolvePodAndContainer(t *testing.T) { + testCases := []struct { + name string + ws *kubefloworgv1beta1.Workspace + requested string + wantPod string + wantContainer string + wantErr error + }{ + { + name: "defaults to the primary 'main' container when none requested", + ws: newWorkspace(testPodName, "main", "istio-proxy"), + requested: "", + wantPod: testPodName, + wantContainer: "main", + }, + { + name: "defaults to 'main' even when it is not the first container", + ws: newWorkspace(testPodName, "istio-proxy", "main"), + requested: "", + wantPod: testPodName, + wantContainer: "main", + }, + { + name: "errors when none requested and no 'main' container exists", + ws: newWorkspace(testPodName, "istio-proxy"), + requested: "", + wantErr: ErrContainerNotFound, + }, + { + name: "returns requested container when it exists", + ws: newWorkspace(testPodName, "main", "istio-proxy"), + requested: "istio-proxy", + wantPod: testPodName, + wantContainer: "istio-proxy", + }, + { + name: "returns requested init container when it exists", + ws: withInitContainers(newWorkspace(testPodName, "main"), "istio-proxy"), + requested: "istio-proxy", + wantPod: testPodName, + wantContainer: "istio-proxy", + }, + { + name: "errors when requested container does not exist", + ws: newWorkspace(testPodName, "main"), + requested: "nope", + wantErr: ErrContainerNotFound, + }, + { + name: "errors when pod name is empty", + ws: newWorkspace("", "main"), + requested: "", + wantErr: ErrPodNotRunning, + }, + { + name: "errors when pod has no containers yet", + ws: newWorkspace(testPodName), + requested: "", + wantErr: ErrContainerNotRunning, + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + repo := newRepo(t, tc.ws, podFromWorkspace(tc.ws)) + pod, container, err := repo.resolvePodAndContainer( + context.Background(), testNamespace, testWorkspace, + &models.LogOptions{Container: tc.requested}, + ) + if tc.wantErr != nil { + if !errors.Is(err, tc.wantErr) { + t.Fatalf("expected error %v, got: %v", tc.wantErr, err) + } + return + } + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if pod != tc.wantPod { + t.Errorf("pod: want %q, got %q", tc.wantPod, pod) + } + if container != tc.wantContainer { + t.Errorf("container: want %q, got %q", tc.wantContainer, container) + } + }) + } +} diff --git a/workspaces/backend/internal/repositories/repositories.go b/workspaces/backend/internal/repositories/repositories.go index be08cf1a3..9a74b8f02 100644 --- a/workspaces/backend/internal/repositories/repositories.go +++ b/workspaces/backend/internal/repositories/repositories.go @@ -17,11 +17,13 @@ limitations under the License. package repositories import ( + "k8s.io/client-go/kubernetes" "sigs.k8s.io/controller-runtime/pkg/client" "github.com/kubeflow/notebooks/workspaces/backend/internal/config" "github.com/kubeflow/notebooks/workspaces/backend/internal/repositories/health_check" "github.com/kubeflow/notebooks/workspaces/backend/internal/repositories/namespaces" + "github.com/kubeflow/notebooks/workspaces/backend/internal/repositories/podlogs" "github.com/kubeflow/notebooks/workspaces/backend/internal/repositories/pvcs" "github.com/kubeflow/notebooks/workspaces/backend/internal/repositories/secrets" "github.com/kubeflow/notebooks/workspaces/backend/internal/repositories/storageclasses" @@ -38,6 +40,7 @@ type Repositories struct { StorageClass *storageclasses.StorageClassRepository Workspace *workspaces.WorkspaceRepository WorkspaceKind *workspacekinds.WorkspaceKindRepository + PodLogs *podlogs.PodLogsRepository } // NewRepositories creates a new Repositories instance from a controller-runtime client. @@ -46,6 +49,7 @@ func NewRepositories( cl client.Client, // configMapClient is a label-filtered cached client for image-source ConfigMaps configMapClient client.Client, + clientset kubernetes.Interface, ) *Repositories { return &Repositories{ HealthCheck: health_check.NewHealthCheckRepository(cfg), @@ -55,5 +59,6 @@ func NewRepositories( StorageClass: storageclasses.NewStorageClassRepository(cfg, cl), Workspace: workspaces.NewWorkspaceRepository(cfg, cl), WorkspaceKind: workspacekinds.NewWorkspaceKindRepository(cfg, cl, configMapClient), + PodLogs: podlogs.NewPodLogsRepository(cfg, cl, clientset), } } diff --git a/workspaces/backend/manifests/kustomize/base/rbac.yaml b/workspaces/backend/manifests/kustomize/base/rbac.yaml index b63273802..45c498292 100644 --- a/workspaces/backend/manifests/kustomize/base/rbac.yaml +++ b/workspaces/backend/manifests/kustomize/base/rbac.yaml @@ -44,6 +44,12 @@ rules: - get - list - watch +- apiGroups: + - "" + resources: + - pods/log + verbs: + - get - apiGroups: - "" resources: diff --git a/workspaces/backend/openapi/docs.go b/workspaces/backend/openapi/docs.go index 3ec938eeb..2dc61e744 100644 --- a/workspaces/backend/openapi/docs.go +++ b/workspaces/backend/openapi/docs.go @@ -1944,6 +1944,105 @@ const docTemplate = `{ } } } + }, + "/workspaces/{namespace}/{name}/podtemplate/logs/batch": { + "get": { + "description": "Returns a point-in-time snapshot of container logs for the workspace pod as a raw text/plain stream proxied directly from the Kubernetes pod logs API.", + "produces": [ + "text/plain" + ], + "tags": [ + "workspaces" + ], + "summary": "Get workspace container logs (batch)", + "operationId": "getWorkspacePodTemplateLogsBatch", + "parameters": [ + { + "type": "string", + "x-example": "kubeflow-user-example-com", + "description": "Namespace of the workspace", + "name": "namespace", + "in": "path", + "required": true + }, + { + "type": "string", + "x-example": "my-workspace", + "description": "Name of the workspace", + "name": "name", + "in": "path", + "required": true + }, + { + "type": "string", + "description": "Target container name. Defaults to the primary (main) container.", + "name": "container", + "in": "query" + }, + { + "type": "integer", + "description": "Number of lines from the end of the log to return. Defaults to 1000.", + "name": "tailLines", + "in": "query" + }, + { + "type": "string", + "description": "Only return logs after this RFC3339 timestamp (e.g. 2026-07-15T10:30:00Z).", + "name": "sinceTime", + "in": "query" + }, + { + "type": "boolean", + "description": "If true, returns logs from the previous terminated container instance.", + "name": "previous", + "in": "query" + } + ], + "responses": { + "200": { + "description": "Raw container log stream (text/plain).", + "schema": { + "type": "string" + } + }, + "400": { + "description": "Bad Request. Container not found, pod not running, container not started, or no previous logs available.", + "schema": { + "$ref": "#/definitions/api.ErrorEnvelope" + } + }, + "401": { + "description": "Unauthorized.", + "schema": { + "$ref": "#/definitions/api.ErrorEnvelope" + } + }, + "403": { + "description": "Forbidden.", + "schema": { + "$ref": "#/definitions/api.ErrorEnvelope" + } + }, + "404": { + "description": "Workspace not found.", + "schema": { + "$ref": "#/definitions/api.ErrorEnvelope" + } + }, + "422": { + "description": "Unprocessable Entity. Validation error.", + "schema": { + "$ref": "#/definitions/api.ErrorEnvelope" + } + }, + "500": { + "description": "Internal server error.", + "schema": { + "$ref": "#/definitions/api.ErrorEnvelope" + } + } + } + } } }, "definitions": { diff --git a/workspaces/backend/openapi/swagger.json b/workspaces/backend/openapi/swagger.json index 153743d3e..a8b8b9712 100644 --- a/workspaces/backend/openapi/swagger.json +++ b/workspaces/backend/openapi/swagger.json @@ -1942,6 +1942,105 @@ } } } + }, + "/workspaces/{namespace}/{name}/podtemplate/logs/batch": { + "get": { + "description": "Returns a point-in-time snapshot of container logs for the workspace pod as a raw text/plain stream proxied directly from the Kubernetes pod logs API.", + "produces": [ + "text/plain" + ], + "tags": [ + "workspaces" + ], + "summary": "Get workspace container logs (batch)", + "operationId": "getWorkspacePodTemplateLogsBatch", + "parameters": [ + { + "type": "string", + "x-example": "kubeflow-user-example-com", + "description": "Namespace of the workspace", + "name": "namespace", + "in": "path", + "required": true + }, + { + "type": "string", + "x-example": "my-workspace", + "description": "Name of the workspace", + "name": "name", + "in": "path", + "required": true + }, + { + "type": "string", + "description": "Target container name. Defaults to the primary (main) container.", + "name": "container", + "in": "query" + }, + { + "type": "integer", + "description": "Number of lines from the end of the log to return. Defaults to 1000.", + "name": "tailLines", + "in": "query" + }, + { + "type": "string", + "description": "Only return logs after this RFC3339 timestamp (e.g. 2026-07-15T10:30:00Z).", + "name": "sinceTime", + "in": "query" + }, + { + "type": "boolean", + "description": "If true, returns logs from the previous terminated container instance.", + "name": "previous", + "in": "query" + } + ], + "responses": { + "200": { + "description": "Raw container log stream (text/plain).", + "schema": { + "type": "string" + } + }, + "400": { + "description": "Bad Request. Container not found, pod not running, container not started, or no previous logs available.", + "schema": { + "$ref": "#/definitions/api.ErrorEnvelope" + } + }, + "401": { + "description": "Unauthorized.", + "schema": { + "$ref": "#/definitions/api.ErrorEnvelope" + } + }, + "403": { + "description": "Forbidden.", + "schema": { + "$ref": "#/definitions/api.ErrorEnvelope" + } + }, + "404": { + "description": "Workspace not found.", + "schema": { + "$ref": "#/definitions/api.ErrorEnvelope" + } + }, + "422": { + "description": "Unprocessable Entity. Validation error.", + "schema": { + "$ref": "#/definitions/api.ErrorEnvelope" + } + }, + "500": { + "description": "Internal server error.", + "schema": { + "$ref": "#/definitions/api.ErrorEnvelope" + } + } + } + } } }, "definitions": {