mirror of
https://github.com/hanzo-ml/trainer.git
synced 2026-09-24 00:31:34 +00:00
* feat(api): add progress types to api Signed-off-by: Rob Bell <[email protected]> * feat: progress server stub Signed-off-by: Rob Bell <[email protected]> * feat: setup tls for server Signed-off-by: Rob Bell <[email protected]> * refactor: pull out middleware Signed-off-by: Rob Bell <[email protected]> * refactor: remove logger message noise from tests Signed-off-by: Rob Bell <[email protected]> * fix: send server error logs to main logger Signed-off-by: Rob Bell <[email protected]> * feat: update the trainer status Signed-off-by: Rob Bell <[email protected]> * feat: use separate client to avoid impacting main reconciler client-side rate limits Signed-off-by: Rob Bell <[email protected]> * feat: configure progress server in helm chart and configmap Signed-off-by: Rob Bell <[email protected]> * fix: prevent race condition Signed-off-by: Rob Bell <[email protected]> * feat: inject progress runtime config Signed-off-by: Rob Bell <[email protected]> * fix: progress client config Signed-off-by: Rob Bell <[email protected]> * fix: update generated config Signed-off-by: Rob Bell <[email protected]> * feat: validate auth tokens in endpoint Signed-off-by: Rob Bell <[email protected]> * test: validate auth tokens in endpoint Signed-off-by: Rob Bell <[email protected]> * refactor: tidy tests to use same format as other tests Signed-off-by: Rob Bell <[email protected]> * fix: look up the cluster issuer from the service account token Different clusters can use different issuers so look up the issuer from the trusted in-cluster service account token. Signed-off-by: Rob Bell <[email protected]> * refactor: rename authn middleware Signed-off-by: Rob Bell <[email protected]> * feat: add authn to status update endpoint Signed-off-by: Rob Bell <[email protected]> * refactor: tidy up reading in-cluster service account token Signed-off-by: Rob Bell <[email protected]> * fix: rbac permissions Signed-off-by: Rob Bell <[email protected]> * fix: better error response Signed-off-by: Rob Bell <[email protected]> * refactor: move kubebuilder annotation higher up Signed-off-by: Rob Bell <[email protected]> * fix: make lastUpdateTime required Signed-off-by: Rob Bell <[email protected]> * chore: fix config tests after rebase Signed-off-by: Rob Bell <[email protected]> * test: update config test with new progress server config Signed-off-by: Rob Bell <[email protected]> * test: add progress tracking e2e test Signed-off-by: Rob Bell <[email protected]> * fix: allow feature gates to be configurable from command line argument So they can be overridden in e2e tests Signed-off-by: Rob Bell <[email protected]> * refactor: reuse shared progress StatusUrl Signed-off-by: Rob Bell <[email protected]> * test: tidy up tests Signed-off-by: Rob Bell <[email protected]> * refactor: allow plugins to use config Signed-off-by: Rob Bell <[email protected]> * refactor: centralise shared constants Signed-off-by: Rob Bell <[email protected]> * fix: disable http2 in progress server tls opts if disabled by config Signed-off-by: Rob Bell <[email protected]> * fix: minor improvements suggested by copilot Signed-off-by: Rob Bell <[email protected]> * chore: rename progress server => status server Signed-off-by: Rob Bell <[email protected]> * chore: update type doc Signed-off-by: Rob Bell <[email protected]> * chore: update name of feature gate Signed-off-by: Rob Bell <[email protected]> * fix: after rebase Signed-off-by: Rob Bell <[email protected]> * chore: fix typo Signed-off-by: Rob Bell <[email protected]> * chore: rename handler Signed-off-by: Rob Bell <[email protected]> * fix: run status server on all replicas Signed-off-by: Rob Bell <[email protected]> * fix: deep copy the rest config Signed-off-by: Rob Bell <[email protected]> * refactor: use ssa Signed-off-by: Rob Bell <[email protected]> * refactor: separate token audience per train job Signed-off-by: Rob Bell <[email protected]> * chore: rename status server Signed-off-by: Rob Bell <[email protected]> * fix: revert adding status pkg to manifest gen paths Signed-off-by: Rob Bell <[email protected]> * chore: update log messages Signed-off-by: Rob Bell <[email protected]> * chore: update api docs Signed-off-by: Rob Bell <[email protected]> * chore: update format Signed-off-by: Rob Bell <[email protected]> * fix: revert inadvertent change Signed-off-by: Rob Bell <[email protected]> * chore: update api name and type docs Signed-off-by: Rob Bell <[email protected]> * refactor: initialise oidc provider in manager startup Signed-off-by: Rob Bell <[email protected]> * fix: allow empty update request Signed-off-by: Rob Bell <[email protected]> * fix: wait for server shut down Signed-off-by: Rob Bell <[email protected]> * fix: handle missing train job Signed-off-by: Rob Bell <[email protected]> * refactor: update env var names Signed-off-by: Rob Bell <[email protected]> * chore: rename packages, plugins, feature gate Signed-off-by: Rob Bell <[email protected]> * refactor: update env var names Signed-off-by: Rob Bell <[email protected]> * chore: regenerate after rebase Signed-off-by: Rob Bell <[email protected]> * chore: fixes after rebase Signed-off-by: Rob Bell <[email protected]> * chore: minor doc updates Signed-off-by: Rob Bell <[email protected]> * chore: rename status server config Signed-off-by: Rob Bell <[email protected]> * chore: replace placeholder config with nil Signed-off-by: Rob Bell <[email protected]> * chore: revert separate runtime utils package Signed-off-by: Rob Bell <[email protected]> * chore: remove unused config from tests Signed-off-by: Rob Bell <[email protected]> --------- Signed-off-by: Rob Bell <[email protected]>
157 lines
4.4 KiB
Go
157 lines
4.4 KiB
Go
/*
|
|
Copyright 2026 The Kubeflow Authors.
|
|
|
|
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 statusserver
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"crypto/tls"
|
|
"encoding/json"
|
|
"io"
|
|
"net/http"
|
|
"net/http/httptest"
|
|
"strings"
|
|
"testing"
|
|
|
|
"github.com/google/go-cmp/cmp"
|
|
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
|
"k8s.io/utils/ptr"
|
|
"sigs.k8s.io/controller-runtime/pkg/client"
|
|
|
|
configapi "github.com/kubeflow/trainer/v2/pkg/apis/config/v1alpha1"
|
|
trainer "github.com/kubeflow/trainer/v2/pkg/apis/trainer/v1alpha1"
|
|
utiltesting "github.com/kubeflow/trainer/v2/pkg/util/testing"
|
|
)
|
|
|
|
type fakeAuthorizer struct {
|
|
authorized bool
|
|
}
|
|
|
|
func (f fakeAuthorizer) Init(_ context.Context) error {
|
|
return nil
|
|
}
|
|
|
|
func (f fakeAuthorizer) Authorize(_ context.Context, _, _, _ string) (bool, error) {
|
|
return f.authorized, nil
|
|
}
|
|
|
|
func newTestServer(t *testing.T, cfg *configapi.StatusServer, authorizer TokenAuthorizer, objs ...client.Object) *httptest.Server {
|
|
t.Helper()
|
|
|
|
fakeClient := utiltesting.NewClientBuilder().
|
|
WithObjects(objs...).
|
|
WithStatusSubresource(objs...).
|
|
Build()
|
|
|
|
srv, err := NewServer(fakeClient, cfg, &tls.Config{}, authorizer)
|
|
if err != nil {
|
|
t.Fatalf("NewServer() error: %v", err)
|
|
}
|
|
|
|
return httptest.NewServer(srv.httpServer.Handler)
|
|
}
|
|
|
|
func TestServerErrorResponses(t *testing.T) {
|
|
cases := map[string]struct {
|
|
url string
|
|
body string
|
|
authorized bool
|
|
wantResponse *metav1.Status
|
|
}{
|
|
"unauthorized fails with 403 unauthorized": {
|
|
url: "/apis/trainer.kubeflow.org/v1alpha1/namespaces/default/trainjobs/test-job/status",
|
|
authorized: false,
|
|
wantResponse: &metav1.Status{
|
|
Status: metav1.StatusFailure,
|
|
Message: "Forbidden",
|
|
Reason: metav1.StatusReasonForbidden,
|
|
Code: http.StatusForbidden,
|
|
},
|
|
},
|
|
"invalid payload fails with 422 unprocessable entity": {
|
|
url: "/apis/trainer.kubeflow.org/v1alpha1/namespaces/default/trainjobs/test-job/status",
|
|
body: `invalid payload`,
|
|
authorized: true,
|
|
wantResponse: &metav1.Status{
|
|
Status: metav1.StatusFailure,
|
|
Message: "Invalid payload",
|
|
Reason: metav1.StatusReasonInvalid,
|
|
Code: http.StatusUnprocessableEntity,
|
|
},
|
|
},
|
|
"oversized payload fails with 413 payload too large error": {
|
|
url: "/apis/trainer.kubeflow.org/v1alpha1/namespaces/default/trainjobs/test-job/status",
|
|
// Generate ~1MB payload (exceeds 64kB limit)
|
|
body: `{"trainerStatus": {"metrics": [` + strings.Repeat(`{"name":"m","value":"0.5"},`, 40000) + `]}}`,
|
|
authorized: true,
|
|
wantResponse: &metav1.Status{
|
|
Status: metav1.StatusFailure,
|
|
Message: "Payload too large",
|
|
Reason: metav1.StatusReasonRequestEntityTooLarge,
|
|
Code: http.StatusRequestEntityTooLarge,
|
|
},
|
|
},
|
|
}
|
|
|
|
for name, tc := range cases {
|
|
t.Run(name, func(t *testing.T) {
|
|
existingTrainJob := &trainer.TrainJob{
|
|
ObjectMeta: metav1.ObjectMeta{
|
|
Name: "test-job",
|
|
Namespace: "default",
|
|
},
|
|
}
|
|
ts := newTestServer(
|
|
t,
|
|
&configapi.StatusServer{Port: ptr.To[int32](8080)},
|
|
fakeAuthorizer{authorized: tc.authorized},
|
|
existingTrainJob,
|
|
)
|
|
defer ts.Close()
|
|
|
|
// Make actual HTTP request
|
|
req, err := http.NewRequest("POST", ts.URL+tc.url, bytes.NewReader([]byte(tc.body)))
|
|
if err != nil {
|
|
t.Fatalf("Failed to create request: %v", err)
|
|
}
|
|
resp, err := http.DefaultClient.Do(req)
|
|
if err != nil {
|
|
t.Fatalf("HTTP POST failed: %v", err)
|
|
}
|
|
t.Cleanup(func() { _ = resp.Body.Close() })
|
|
|
|
if resp.StatusCode != int(tc.wantResponse.Code) {
|
|
t.Errorf("status = %v, want %v", resp.StatusCode, tc.wantResponse.Code)
|
|
}
|
|
|
|
body, err := io.ReadAll(resp.Body)
|
|
if err != nil {
|
|
t.Fatalf("Failed to read response body: %v", err)
|
|
}
|
|
|
|
var got metav1.Status
|
|
if err := json.Unmarshal(body, &got); err != nil {
|
|
t.Fatalf("Failed to unmarshal response: %v", err)
|
|
}
|
|
|
|
if diff := cmp.Diff(tc.wantResponse, &got); diff != "" {
|
|
t.Errorf("response mismatch (-want +got):\n%s", diff)
|
|
}
|
|
})
|
|
}
|
|
}
|