refactor: move Ochre serving lifecycle to the Ochre domain

handlers/bedrock becomes domains/ochre. Ochre is the platform service;
Bedrock names only its AWS API surface. Persisted and observable
identifiers keep their bedrock names.
This commit is contained in:
joshsiv-mulga
2026-10-06 16:40:38 +10:00
parent f329ec3f32
commit 2c467988d3
44 changed files with 176 additions and 176 deletions
+20 -20
View File
@@ -9,7 +9,7 @@ import (
"strconv"
"time"
handlers_bedrock "github.com/mulgadc/spinifex/spinifex/handlers/bedrock"
"github.com/mulgadc/spinifex/spinifex/domains/ochre"
"github.com/pterm/pterm"
"github.com/spf13/cobra"
)
@@ -118,18 +118,18 @@ func realEndpointWaitClock() endpointWaitClock {
// waitForEndpointReady polls Describe until the endpoint is READY, has gone
// back to ABSENT, or the timeout expires, and returns the record it settled
// on plus how long that took.
func waitForEndpointReady(ctx context.Context, svc handlers_bedrock.EndpointService, modelID string,
timeout time.Duration, clock endpointWaitClock) (handlers_bedrock.EndpointRecord, time.Duration, error) {
func waitForEndpointReady(ctx context.Context, svc ochre.EndpointService, modelID string,
timeout time.Duration, clock endpointWaitClock) (ochre.EndpointRecord, time.Duration, error) {
start := clock.now()
for {
out, err := svc.Describe(ctx, &handlers_bedrock.DescribeEndpointInput{ModelID: modelID}, awsidentifiers.GlobalAccountID)
out, err := svc.Describe(ctx, &ochre.DescribeEndpointInput{ModelID: modelID}, awsidentifiers.GlobalAccountID)
if err != nil {
return handlers_bedrock.EndpointRecord{}, clock.now().Sub(start), err
return ochre.EndpointRecord{}, clock.now().Sub(start), err
}
switch out.Endpoint.State {
case handlers_bedrock.StateReady:
case ochre.StateReady:
return out.Endpoint, clock.now().Sub(start), nil
case handlers_bedrock.StateAbsent:
case ochre.StateAbsent:
return out.Endpoint, clock.now().Sub(start), errEndpointLaunchAborted
}
@@ -144,7 +144,7 @@ func waitForEndpointReady(ctx context.Context, svc handlers_bedrock.EndpointServ
// formatEndpointRecord renders one record as aligned key/value lines, omitting
// fields that are only set once a launch has progressed far enough to have them.
func formatEndpointRecord(rec handlers_bedrock.EndpointRecord) string {
func formatEndpointRecord(rec ochre.EndpointRecord) string {
rows := [][2]string{
{"Model ID", rec.ModelID},
}
@@ -194,8 +194,8 @@ func formatEndpointRecord(rec handlers_bedrock.EndpointRecord) string {
// "Idle for" is measured from the record's LastActive, which falls back to
// ReadyAt, so an endpoint that has been quiet since launch reads as idle since
// launch rather than since the zero time.
func reclaimRows(rec handlers_bedrock.EndpointRecord) [][2]string {
if rec.State != handlers_bedrock.StateReady {
func reclaimRows(rec ochre.EndpointRecord) [][2]string {
if rec.State != ochre.StateReady {
return nil
}
rows := [][2]string{{"In flight", strconv.Itoa(rec.InFlight)}}
@@ -215,8 +215,8 @@ func reclaimRows(rec handlers_bedrock.EndpointRecord) [][2]string {
// ACCOUNT and PINNED distinguish a pinned, account-scoped endpoint from a
// shared platform one — List itself now returns every account's records, not
// just the shared platform account's.
func listEndpointsOutput(ctx context.Context, svc handlers_bedrock.EndpointService) (string, error) {
out, err := svc.List(ctx, &handlers_bedrock.ListEndpointsInput{}, awsidentifiers.GlobalAccountID)
func listEndpointsOutput(ctx context.Context, svc ochre.EndpointService) (string, error) {
out, err := svc.List(ctx, &ochre.ListEndpointsInput{}, awsidentifiers.GlobalAccountID)
if err != nil {
return "", err
}
@@ -237,9 +237,9 @@ func listEndpointsOutput(ctx context.Context, svc handlers_bedrock.EndpointServi
// runEnsureEndpoint is the testable core of 'ochre endpoint ensure': request
// the endpoint, then optionally wait for it. Returns the message to print.
func runEnsureEndpoint(ctx context.Context, svc handlers_bedrock.EndpointService, modelID string,
func runEnsureEndpoint(ctx context.Context, svc ochre.EndpointService, modelID string,
wait bool, timeout time.Duration, clock endpointWaitClock) (string, error) {
out, err := svc.Ensure(ctx, &handlers_bedrock.EnsureEndpointInput{ModelID: modelID}, awsidentifiers.GlobalAccountID)
out, err := svc.Ensure(ctx, &ochre.EnsureEndpointInput{ModelID: modelID}, awsidentifiers.GlobalAccountID)
if err != nil {
return "", err
}
@@ -249,7 +249,7 @@ func runEnsureEndpoint(ctx context.Context, svc handlers_bedrock.EndpointService
// Already READY before any polling means this was a warm request, not a
// cold start, so reporting an elapsed time would be misleading.
if out.Endpoint.State == handlers_bedrock.StateReady {
if out.Endpoint.State == ochre.StateReady {
return fmt.Sprintf("Endpoint for %s was already READY.\n\n%s", modelID, formatEndpointRecord(out.Endpoint)), nil
}
@@ -263,12 +263,12 @@ func runEnsureEndpoint(ctx context.Context, svc handlers_bedrock.EndpointService
// endpointServiceFn indirects the NATS-backed client so the Run functions'
// connect/exit control flow can be tested without a live daemon.
var endpointServiceFn = func() (handlers_bedrock.EndpointService, func(), error) {
var endpointServiceFn = func() (ochre.EndpointService, func(), error) {
_, nc, err := loadConfigAndConnectFn()
if err != nil {
return nil, nil, err
}
return handlers_bedrock.NewNATSEndpointService(nc), nc.Close, nil
return ochre.NewNATSEndpointService(nc), nc.Close, nil
}
func runOchreEndpointEnsure(cmd *cobra.Command, _ []string) {
@@ -305,7 +305,7 @@ func runOchreEndpointDescribe(cmd *cobra.Command, _ []string) {
}
defer closeFn()
out, err := svc.Describe(context.Background(), &handlers_bedrock.DescribeEndpointInput{ModelID: modelID, AccountID: accountID}, awsidentifiers.GlobalAccountID)
out, err := svc.Describe(context.Background(), &ochre.DescribeEndpointInput{ModelID: modelID, AccountID: accountID}, awsidentifiers.GlobalAccountID)
if err != nil {
fmt.Fprintf(os.Stderr, "%v\n", err)
ochreExit(1)
@@ -336,8 +336,8 @@ func runOchreEndpointList(_ *cobra.Command, _ []string) {
// tears the endpoint down and returns an honest message — a no-op that found
// no record must not claim a teardown, and it points the operator at --account
// so a pinned, account-scoped record they can see in 'list' is reachable.
func deleteEndpointOutput(ctx context.Context, svc handlers_bedrock.EndpointService, modelID, accountID string) (string, error) {
out, err := svc.Delete(ctx, &handlers_bedrock.DeleteEndpointInput{ModelID: modelID, AccountID: accountID}, awsidentifiers.GlobalAccountID)
func deleteEndpointOutput(ctx context.Context, svc ochre.EndpointService, modelID, accountID string) (string, error) {
out, err := svc.Delete(ctx, &ochre.DeleteEndpointInput{ModelID: modelID, AccountID: accountID}, awsidentifiers.GlobalAccountID)
if err != nil {
return "", err
}
+66 -66
View File
@@ -7,36 +7,36 @@ import (
"testing"
"time"
handlers_bedrock "github.com/mulgadc/spinifex/spinifex/handlers/bedrock"
"github.com/mulgadc/spinifex/spinifex/domains/ochre"
"github.com/stretchr/testify/require"
)
// fakeEndpointService serves scripted Describe responses so the wait loop can
// be exercised without a daemon, a VM or a real cold start.
type fakeEndpointService struct {
ensure handlers_bedrock.EndpointRecord
ensure ochre.EndpointRecord
ensureErr error
describes []handlers_bedrock.EndpointRecord
describes []ochre.EndpointRecord
describeErr error
list []handlers_bedrock.EndpointRecord
list []ochre.EndpointRecord
listErr error
deleteErr error
deleteRemoved bool
describeCalls int
deleteCalls int
describeInputs []*handlers_bedrock.DescribeEndpointInput
deleteInputs []*handlers_bedrock.DeleteEndpointInput
describeInputs []*ochre.DescribeEndpointInput
deleteInputs []*ochre.DeleteEndpointInput
}
func (f *fakeEndpointService) Ensure(_ context.Context, _ *handlers_bedrock.EnsureEndpointInput, _ string) (*handlers_bedrock.EnsureEndpointOutput, error) {
func (f *fakeEndpointService) Ensure(_ context.Context, _ *ochre.EnsureEndpointInput, _ string) (*ochre.EnsureEndpointOutput, error) {
if f.ensureErr != nil {
return nil, f.ensureErr
}
return &handlers_bedrock.EnsureEndpointOutput{Endpoint: f.ensure}, nil
return &ochre.EnsureEndpointOutput{Endpoint: f.ensure}, nil
}
func (f *fakeEndpointService) Describe(_ context.Context, in *handlers_bedrock.DescribeEndpointInput, _ string) (*handlers_bedrock.DescribeEndpointOutput, error) {
func (f *fakeEndpointService) Describe(_ context.Context, in *ochre.DescribeEndpointInput, _ string) (*ochre.DescribeEndpointOutput, error) {
f.describeInputs = append(f.describeInputs, in)
if f.describeErr != nil {
return nil, f.describeErr
@@ -45,26 +45,26 @@ func (f *fakeEndpointService) Describe(_ context.Context, in *handlers_bedrock.D
// terminal state keeps polling rather than running off the end.
idx := min(f.describeCalls, len(f.describes)-1)
f.describeCalls++
return &handlers_bedrock.DescribeEndpointOutput{Endpoint: f.describes[idx]}, nil
return &ochre.DescribeEndpointOutput{Endpoint: f.describes[idx]}, nil
}
func (f *fakeEndpointService) List(_ context.Context, _ *handlers_bedrock.ListEndpointsInput, _ string) (*handlers_bedrock.ListEndpointsOutput, error) {
func (f *fakeEndpointService) List(_ context.Context, _ *ochre.ListEndpointsInput, _ string) (*ochre.ListEndpointsOutput, error) {
if f.listErr != nil {
return nil, f.listErr
}
return &handlers_bedrock.ListEndpointsOutput{Endpoints: f.list}, nil
return &ochre.ListEndpointsOutput{Endpoints: f.list}, nil
}
func (f *fakeEndpointService) Delete(_ context.Context, in *handlers_bedrock.DeleteEndpointInput, _ string) (*handlers_bedrock.DeleteEndpointOutput, error) {
func (f *fakeEndpointService) Delete(_ context.Context, in *ochre.DeleteEndpointInput, _ string) (*ochre.DeleteEndpointOutput, error) {
f.deleteCalls++
f.deleteInputs = append(f.deleteInputs, in)
if f.deleteErr != nil {
return nil, f.deleteErr
}
return &handlers_bedrock.DeleteEndpointOutput{Removed: f.deleteRemoved}, nil
return &ochre.DeleteEndpointOutput{Removed: f.deleteRemoved}, nil
}
var _ handlers_bedrock.EndpointService = (*fakeEndpointService)(nil)
var _ ochre.EndpointService = (*fakeEndpointService)(nil)
// fakeClock advances a virtual now by whatever the loop sleeps, so a
// multi-minute wait costs nothing and elapsed times are exact.
@@ -78,20 +78,20 @@ func fakeClock(now *time.Time) endpointWaitClock {
const testModelID = "meta.llama3-2-1b-instruct-v1:0"
// testAccountID stands in for a caller's real (non-Global) account, mirroring
// handlers_bedrock's own testAccountID.
// domains/ochre's own testAccountID.
const testAccountID = "111111111111"
func TestWaitForEndpointReady_ReachesReadyAndReportsElapsed(t *testing.T) {
svc := &fakeEndpointService{describes: []handlers_bedrock.EndpointRecord{
{ModelID: testModelID, State: handlers_bedrock.StateStarting},
{ModelID: testModelID, State: handlers_bedrock.StateStarting},
{ModelID: testModelID, State: handlers_bedrock.StateReady, BaseURL: "http://10.0.0.5:8000"},
svc := &fakeEndpointService{describes: []ochre.EndpointRecord{
{ModelID: testModelID, State: ochre.StateStarting},
{ModelID: testModelID, State: ochre.StateStarting},
{ModelID: testModelID, State: ochre.StateReady, BaseURL: "http://10.0.0.5:8000"},
}}
now := time.Unix(0, 0).UTC()
rec, elapsed, err := waitForEndpointReady(context.Background(), svc, testModelID, time.Minute, fakeClock(&now))
require.NoError(t, err)
require.Equal(t, handlers_bedrock.StateReady, rec.State)
require.Equal(t, ochre.StateReady, rec.State)
require.Equal(t, "http://10.0.0.5:8000", rec.BaseURL)
// Two sleeps between three polls.
require.Equal(t, 2*endpointPollInterval, elapsed)
@@ -100,9 +100,9 @@ func TestWaitForEndpointReady_ReachesReadyAndReportsElapsed(t *testing.T) {
// A failed launch deletes the record, so ABSENT after STARTING is the failure
// signal and must not be mistaken for "not started yet".
func TestWaitForEndpointReady_AbsentAfterStartingIsAbort(t *testing.T) {
svc := &fakeEndpointService{describes: []handlers_bedrock.EndpointRecord{
{ModelID: testModelID, State: handlers_bedrock.StateStarting},
{ModelID: testModelID, State: handlers_bedrock.StateAbsent},
svc := &fakeEndpointService{describes: []ochre.EndpointRecord{
{ModelID: testModelID, State: ochre.StateStarting},
{ModelID: testModelID, State: ochre.StateAbsent},
}}
now := time.Unix(0, 0).UTC()
@@ -111,28 +111,28 @@ func TestWaitForEndpointReady_AbsentAfterStartingIsAbort(t *testing.T) {
}
func TestWaitForEndpointReady_TimesOutWhileStillStarting(t *testing.T) {
svc := &fakeEndpointService{describes: []handlers_bedrock.EndpointRecord{
{ModelID: testModelID, State: handlers_bedrock.StateStarting},
svc := &fakeEndpointService{describes: []ochre.EndpointRecord{
{ModelID: testModelID, State: ochre.StateStarting},
}}
now := time.Unix(0, 0).UTC()
rec, elapsed, err := waitForEndpointReady(context.Background(), svc, testModelID, 10*time.Second, fakeClock(&now))
require.ErrorIs(t, err, errEndpointWaitTimeout)
require.Equal(t, handlers_bedrock.StateStarting, rec.State)
require.Equal(t, ochre.StateStarting, rec.State)
require.GreaterOrEqual(t, elapsed, 10*time.Second)
}
// A zero timeout must still report the endpoint's real state, not an empty
// record, because the deadline is only checked after a Describe.
func TestWaitForEndpointReady_ZeroTimeoutStillDescribesOnce(t *testing.T) {
svc := &fakeEndpointService{describes: []handlers_bedrock.EndpointRecord{
{ModelID: testModelID, State: handlers_bedrock.StateStarting},
svc := &fakeEndpointService{describes: []ochre.EndpointRecord{
{ModelID: testModelID, State: ochre.StateStarting},
}}
now := time.Unix(0, 0).UTC()
rec, _, err := waitForEndpointReady(context.Background(), svc, testModelID, 0, fakeClock(&now))
require.ErrorIs(t, err, errEndpointWaitTimeout)
require.Equal(t, handlers_bedrock.StateStarting, rec.State)
require.Equal(t, ochre.StateStarting, rec.State)
require.Equal(t, 1, svc.describeCalls)
}
@@ -145,7 +145,7 @@ func TestWaitForEndpointReady_DescribeErrorSurfaces(t *testing.T) {
}
func TestRunEnsureEndpoint_NoWaitReportsStarting(t *testing.T) {
svc := &fakeEndpointService{ensure: handlers_bedrock.EndpointRecord{ModelID: testModelID, State: handlers_bedrock.StateStarting}}
svc := &fakeEndpointService{ensure: ochre.EndpointRecord{ModelID: testModelID, State: ochre.StateStarting}}
now := time.Unix(0, 0).UTC()
msg, err := runEnsureEndpoint(context.Background(), svc, testModelID, false, time.Minute, fakeClock(&now))
@@ -156,10 +156,10 @@ func TestRunEnsureEndpoint_NoWaitReportsStarting(t *testing.T) {
func TestRunEnsureEndpoint_WaitReportsColdStart(t *testing.T) {
svc := &fakeEndpointService{
ensure: handlers_bedrock.EndpointRecord{ModelID: testModelID, State: handlers_bedrock.StateStarting},
describes: []handlers_bedrock.EndpointRecord{
{ModelID: testModelID, State: handlers_bedrock.StateStarting},
{ModelID: testModelID, State: handlers_bedrock.StateReady},
ensure: ochre.EndpointRecord{ModelID: testModelID, State: ochre.StateStarting},
describes: []ochre.EndpointRecord{
{ModelID: testModelID, State: ochre.StateStarting},
{ModelID: testModelID, State: ochre.StateReady},
},
}
now := time.Unix(0, 0).UTC()
@@ -172,7 +172,7 @@ func TestRunEnsureEndpoint_WaitReportsColdStart(t *testing.T) {
// An endpoint already READY when ensure returns was a warm request; reporting
// an elapsed cold start for it would be misleading.
func TestRunEnsureEndpoint_AlreadyReadyDoesNotReportElapsed(t *testing.T) {
svc := &fakeEndpointService{ensure: handlers_bedrock.EndpointRecord{ModelID: testModelID, State: handlers_bedrock.StateReady}}
svc := &fakeEndpointService{ensure: ochre.EndpointRecord{ModelID: testModelID, State: ochre.StateReady}}
now := time.Unix(0, 0).UTC()
msg, err := runEnsureEndpoint(context.Background(), svc, testModelID, true, time.Minute, fakeClock(&now))
@@ -194,8 +194,8 @@ func TestRunEnsureEndpoint_EnsureErrorSurfaces(t *testing.T) {
// tell a slow launch from a stuck one.
func TestRunEnsureEndpoint_WaitTimeoutIncludesRecord(t *testing.T) {
svc := &fakeEndpointService{
ensure: handlers_bedrock.EndpointRecord{ModelID: testModelID, State: handlers_bedrock.StateStarting},
describes: []handlers_bedrock.EndpointRecord{{ModelID: testModelID, State: handlers_bedrock.StateStarting}},
ensure: ochre.EndpointRecord{ModelID: testModelID, State: ochre.StateStarting},
describes: []ochre.EndpointRecord{{ModelID: testModelID, State: ochre.StateStarting}},
}
now := time.Unix(0, 0).UTC()
@@ -212,9 +212,9 @@ func TestListEndpointsOutput_NoEndpoints(t *testing.T) {
}
func TestListEndpointsOutput_ListsEndpoints(t *testing.T) {
svc := &fakeEndpointService{list: []handlers_bedrock.EndpointRecord{
{ModelID: testModelID, State: handlers_bedrock.StateReady, InstanceID: "i-abc", BaseURL: "http://10.0.0.5:8000"},
{ModelID: "meta.llama3-2-3b-instruct-v1:0", State: handlers_bedrock.StateStarting},
svc := &fakeEndpointService{list: []ochre.EndpointRecord{
{ModelID: testModelID, State: ochre.StateReady, InstanceID: "i-abc", BaseURL: "http://10.0.0.5:8000"},
{ModelID: "meta.llama3-2-3b-instruct-v1:0", State: ochre.StateStarting},
}}
msg, err := listEndpointsOutput(context.Background(), svc)
@@ -229,9 +229,9 @@ func TestListEndpointsOutput_ListsEndpoints(t *testing.T) {
// indicator, while a bare GlobalAccountID endpoint keeps listing unchanged
// (its ACCOUNT cell just reads GlobalAccountID, PINNED empty).
func TestListEndpointsOutput_ShowsAccountAndPinnedColumns(t *testing.T) {
svc := &fakeEndpointService{list: []handlers_bedrock.EndpointRecord{
{ModelID: testModelID, State: handlers_bedrock.StateReady, AccountID: "000000000001"},
{ModelID: "meta.llama3-2-3b-instruct-v1:0", State: handlers_bedrock.StateReady, AccountID: testAccountID, Pinned: true},
svc := &fakeEndpointService{list: []ochre.EndpointRecord{
{ModelID: testModelID, State: ochre.StateReady, AccountID: "000000000001"},
{ModelID: "meta.llama3-2-3b-instruct-v1:0", State: ochre.StateReady, AccountID: testAccountID, Pinned: true},
}}
msg, err := listEndpointsOutput(context.Background(), svc)
@@ -250,9 +250,9 @@ func TestListEndpointsOutput_ErrorSurfaces(t *testing.T) {
func TestFormatEndpointRecord_OmitsUnsetFieldsAndDerivesStartup(t *testing.T) {
created := time.Unix(1000, 0).UTC()
rec := handlers_bedrock.EndpointRecord{
rec := ochre.EndpointRecord{
ModelID: testModelID,
State: handlers_bedrock.StateReady,
State: ochre.StateReady,
CreatedAt: created,
ReadyAt: created.Add(97 * time.Second),
BaseURL: "http://10.0.0.5:8000",
@@ -266,18 +266,18 @@ func TestFormatEndpointRecord_OmitsUnsetFieldsAndDerivesStartup(t *testing.T) {
}
func TestFormatEndpointRecord_AbsentIsMinimal(t *testing.T) {
out := formatEndpointRecord(handlers_bedrock.EndpointRecord{ModelID: testModelID, State: handlers_bedrock.StateAbsent})
out := formatEndpointRecord(ochre.EndpointRecord{ModelID: testModelID, State: ochre.StateAbsent})
require.Contains(t, out, "ABSENT")
require.Equal(t, 2, strings.Count(strings.TrimSpace(out), "\n")+1)
}
// withEndpointService swaps endpointServiceFn for one returning svc, so a Run
// wrapper's real connect/validate/exit control flow runs without a daemon.
func withEndpointService(t *testing.T, svc handlers_bedrock.EndpointService, connErr error) {
func withEndpointService(t *testing.T, svc ochre.EndpointService, connErr error) {
t.Helper()
orig := endpointServiceFn
t.Cleanup(func() { endpointServiceFn = orig })
endpointServiceFn = func() (handlers_bedrock.EndpointService, func(), error) {
endpointServiceFn = func() (ochre.EndpointService, func(), error) {
if connErr != nil {
return nil, nil, connErr
}
@@ -287,7 +287,7 @@ func withEndpointService(t *testing.T, svc handlers_bedrock.EndpointService, con
func TestRunOchreEndpointEnsure_PrintsRecord(t *testing.T) {
withEndpointService(t, &fakeEndpointService{
ensure: handlers_bedrock.EndpointRecord{ModelID: testModelID, State: handlers_bedrock.StateStarting},
ensure: ochre.EndpointRecord{ModelID: testModelID, State: ochre.StateStarting},
}, nil)
cmd := *ochreEndpointEnsureCmd
@@ -322,8 +322,8 @@ func TestRunOchreEndpointEnsure_ServiceErrorExits1(t *testing.T) {
}
func TestRunOchreEndpointDescribe_PrintsRecord(t *testing.T) {
withEndpointService(t, &fakeEndpointService{describes: []handlers_bedrock.EndpointRecord{
{ModelID: testModelID, State: handlers_bedrock.StateReady, InstanceID: "i-abc"},
withEndpointService(t, &fakeEndpointService{describes: []ochre.EndpointRecord{
{ModelID: testModelID, State: ochre.StateReady, InstanceID: "i-abc"},
}}, nil)
cmd := *ochreEndpointDescribeCmd
@@ -342,8 +342,8 @@ func TestRunOchreEndpointDescribe_PrintsRecord(t *testing.T) {
// still send an empty AccountID — resolveAccountID's own GlobalAccountID
// fallback, unchanged from before --account existed.
func TestRunOchreEndpointDescribe_DefaultAccountIsEmpty(t *testing.T) {
svc := &fakeEndpointService{describes: []handlers_bedrock.EndpointRecord{
{ModelID: testModelID, State: handlers_bedrock.StateReady},
svc := &fakeEndpointService{describes: []ochre.EndpointRecord{
{ModelID: testModelID, State: ochre.StateReady},
}}
withEndpointService(t, svc, nil)
@@ -362,8 +362,8 @@ func TestRunOchreEndpointDescribe_DefaultAccountIsEmpty(t *testing.T) {
// operator sees a pinned, account-scoped endpoint the GlobalAccountID lookup
// would otherwise report ABSENT.
func TestRunOchreEndpointDescribe_AccountFlagScopesLookup(t *testing.T) {
svc := &fakeEndpointService{describes: []handlers_bedrock.EndpointRecord{
{ModelID: testModelID, State: handlers_bedrock.StateReady, AccountID: testAccountID, Pinned: true},
svc := &fakeEndpointService{describes: []ochre.EndpointRecord{
{ModelID: testModelID, State: ochre.StateReady, AccountID: testAccountID, Pinned: true},
}}
withEndpointService(t, svc, nil)
@@ -392,8 +392,8 @@ func TestRunOchreEndpointDescribe_ErrorExits1(t *testing.T) {
}
func TestRunOchreEndpointList_PrintsTable(t *testing.T) {
withEndpointService(t, &fakeEndpointService{list: []handlers_bedrock.EndpointRecord{
{ModelID: testModelID, State: handlers_bedrock.StateReady},
withEndpointService(t, &fakeEndpointService{list: []ochre.EndpointRecord{
{ModelID: testModelID, State: ochre.StateReady},
}}, nil)
var out string
@@ -480,9 +480,9 @@ func TestRunOchreEndpointDelete_ErrorExits1(t *testing.T) {
// endpoint reaped" or "why was it not", so a READY record must show them.
func TestFormatEndpointRecord_ShowsReclaimInputs(t *testing.T) {
ready := time.Now().UTC().Add(-30 * time.Minute)
out := formatEndpointRecord(handlers_bedrock.EndpointRecord{
out := formatEndpointRecord(ochre.EndpointRecord{
ModelID: testModelID,
State: handlers_bedrock.StateReady,
State: ochre.StateReady,
ReadyAt: ready,
LastActiveAt: ready.Add(20 * time.Minute),
InFlight: 3,
@@ -497,9 +497,9 @@ func TestFormatEndpointRecord_ShowsReclaimInputs(t *testing.T) {
// An endpoint quiet since launch is idle since launch, not since the zero
// time: the fallback is what keeps the reported figure meaningful.
func TestFormatEndpointRecord_NeverActiveFallsBackToReadyAt(t *testing.T) {
out := formatEndpointRecord(handlers_bedrock.EndpointRecord{
out := formatEndpointRecord(ochre.EndpointRecord{
ModelID: testModelID,
State: handlers_bedrock.StateReady,
State: ochre.StateReady,
ReadyAt: time.Now().UTC().Add(-90 * time.Second),
})
@@ -509,9 +509,9 @@ func TestFormatEndpointRecord_NeverActiveFallsBackToReadyAt(t *testing.T) {
}
func TestFormatEndpointRecord_SurfacesPinnedAndScrapeFailures(t *testing.T) {
out := formatEndpointRecord(handlers_bedrock.EndpointRecord{
out := formatEndpointRecord(ochre.EndpointRecord{
ModelID: testModelID,
State: handlers_bedrock.StateReady,
State: ochre.StateReady,
ReadyAt: time.Now().UTC().Add(-time.Hour),
Pinned: true,
ScrapeFailures: 2,
@@ -525,9 +525,9 @@ func TestFormatEndpointRecord_SurfacesPinnedAndScrapeFailures(t *testing.T) {
// Only READY endpoints are swept, so reclaim rows on any other state would be
// reporting a decision nothing is making.
func TestFormatEndpointRecord_NoReclaimRowsWhenNotReady(t *testing.T) {
out := formatEndpointRecord(handlers_bedrock.EndpointRecord{
out := formatEndpointRecord(ochre.EndpointRecord{
ModelID: testModelID,
State: handlers_bedrock.StateStarting,
State: ochre.StateStarting,
})
require.NotContains(t, out, "In flight")
+5 -5
View File
@@ -3,7 +3,7 @@ package accountteardown
import (
"context"
handlers_bedrock "github.com/mulgadc/spinifex/spinifex/handlers/bedrock"
"github.com/mulgadc/spinifex/spinifex/domains/ochre"
"github.com/nats-io/nats.go"
)
@@ -13,11 +13,11 @@ import (
// and a weights volume, and only its own delete releases all three. Terminating
// its instance first would strand the other two with no record naming them.
func BedrockReapers(nc *nats.Conn) []Reaper {
return []Reaper{&bedrockEndpointReaper{svc: handlers_bedrock.NewNATSEndpointService(nc)}}
return []Reaper{&bedrockEndpointReaper{svc: ochre.NewNATSEndpointService(nc)}}
}
type bedrockEndpointReaper struct {
svc handlers_bedrock.EndpointService
svc ochre.EndpointService
}
func (r *bedrockEndpointReaper) Kind() string { return "bedrock-endpoint" }
@@ -28,7 +28,7 @@ func (r *bedrockEndpointReaper) Stage() Stage { return StageCompute }
// report — and then delete — endpoints belonging to other tenants and to the
// shared global account.
func (r *bedrockEndpointReaper) List(ctx context.Context, accountID string) ([]Resource, error) {
out, err := r.svc.List(ctx, &handlers_bedrock.ListEndpointsInput{}, accountID)
out, err := r.svc.List(ctx, &ochre.ListEndpointsInput{}, accountID)
if err != nil {
return nil, err
}
@@ -50,7 +50,7 @@ func (r *bedrockEndpointReaper) List(ctx context.Context, accountID string) ([]R
// Delete names the account explicitly: an empty AccountID resolves to the
// global one, which would delete the shared endpoint instead of the tenant's.
func (r *bedrockEndpointReaper) Delete(ctx context.Context, accountID string, resource Resource, _ bool) error {
_, err := r.svc.Delete(ctx, &handlers_bedrock.DeleteEndpointInput{
_, err := r.svc.Delete(ctx, &ochre.DeleteEndpointInput{
ModelID: resource.ID,
AccountID: accountID,
}, accountID)
@@ -8,40 +8,40 @@ import (
"errors"
"testing"
handlers_bedrock "github.com/mulgadc/spinifex/spinifex/handlers/bedrock"
"github.com/mulgadc/spinifex/spinifex/domains/ochre"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
type fakeEndpoints struct {
handlers_bedrock.EndpointService
ochre.EndpointService
records []handlers_bedrock.EndpointRecord
deleted []handlers_bedrock.DeleteEndpointInput
records []ochre.EndpointRecord
deleted []ochre.DeleteEndpointInput
err error
}
func (f *fakeEndpoints) List(_ context.Context, _ *handlers_bedrock.ListEndpointsInput, _ string) (*handlers_bedrock.ListEndpointsOutput, error) {
return &handlers_bedrock.ListEndpointsOutput{Endpoints: f.records}, nil
func (f *fakeEndpoints) List(_ context.Context, _ *ochre.ListEndpointsInput, _ string) (*ochre.ListEndpointsOutput, error) {
return &ochre.ListEndpointsOutput{Endpoints: f.records}, nil
}
func (f *fakeEndpoints) Delete(_ context.Context, in *handlers_bedrock.DeleteEndpointInput, _ string) (*handlers_bedrock.DeleteEndpointOutput, error) {
func (f *fakeEndpoints) Delete(_ context.Context, in *ochre.DeleteEndpointInput, _ string) (*ochre.DeleteEndpointOutput, error) {
if f.err != nil {
return nil, f.err
}
f.deleted = append(f.deleted, *in)
return &handlers_bedrock.DeleteEndpointOutput{}, nil
return &ochre.DeleteEndpointOutput{}, nil
}
// Bedrock's List answers with every endpoint in the cluster whatever account
// it is called for, so the filter is the reaper's own responsibility: without
// it a tenant teardown would delete the shared endpoint and other tenants'.
func TestBedrockEndpointReaperListsOnlyItsOwnAccount(t *testing.T) {
svc := &fakeEndpoints{records: []handlers_bedrock.EndpointRecord{
{AccountID: "000000000002", ModelID: "mistral-7b", State: handlers_bedrock.StateReady},
{AccountID: "000000000003", ModelID: "llama-3", State: handlers_bedrock.StateReady},
{AccountID: "global", ModelID: "shared", State: handlers_bedrock.StateReady},
{AccountID: "000000000002", ModelID: "", State: handlers_bedrock.StateStarting},
svc := &fakeEndpoints{records: []ochre.EndpointRecord{
{AccountID: "000000000002", ModelID: "mistral-7b", State: ochre.StateReady},
{AccountID: "000000000003", ModelID: "llama-3", State: ochre.StateReady},
{AccountID: "global", ModelID: "shared", State: ochre.StateReady},
{AccountID: "000000000002", ModelID: "", State: ochre.StateStarting},
}}
reaper := &bedrockEndpointReaper{svc: svc}
+1 -1
View File
@@ -406,7 +406,7 @@ type OchreVectorConfig struct {
Enabled bool `json:"Enabled" mapstructure:"enabled"`
// EmbeddingModel documents this deployment's served embedding model id;
// each index still pins its own model id at CreateIndex time. Any
// self-host model id resolves via handlers_bedrock.DynamicEndpointResolver.
// self-host model id resolves via ochre.DynamicEndpointResolver.
EmbeddingModel string `json:"EmbeddingModel" mapstructure:"embedding_model"`
// RerankModel is the served cross-encoder rerank model id, resolved the
// same way EmbeddingModel is. Empty disables reranking entirely: Query
+6 -6
View File
@@ -4,8 +4,8 @@ import (
"log/slog"
"github.com/mulgadc/spinifex/spinifex/domains/network/systemvpc"
"github.com/mulgadc/spinifex/spinifex/domains/ochre"
gateway_bedrock "github.com/mulgadc/spinifex/spinifex/gateway/bedrock"
handlers_bedrock "github.com/mulgadc/spinifex/spinifex/handlers/bedrock"
handlers_rds "github.com/mulgadc/spinifex/spinifex/handlers/rds"
"github.com/mulgadc/spinifex/spinifex/network/host"
"github.com/mulgadc/spinifex/spinifex/runtime/compute/gpu"
@@ -16,7 +16,7 @@ import (
// plumbing mirrors buildRDSLaunchDeps exactly (same underlying EC2 handler
// set), plus the weights snapshot resolver the RDS launcher has no equivalent
// of.
func (d *Daemon) buildBedrockLaunchDeps() handlers_bedrock.LaunchDeps {
func (d *Daemon) buildBedrockLaunchDeps() ochre.LaunchDeps {
js, err := jetstream.New(d.natsConn)
var weights = gateway_bedrock.NoopWeightsResolver
if err != nil {
@@ -25,7 +25,7 @@ func (d *Daemon) buildBedrockLaunchDeps() handlers_bedrock.LaunchDeps {
weights = gateway_bedrock.NewWeightsStore(js)
}
return handlers_bedrock.LaunchDeps{
return ochre.LaunchDeps{
Config: d.config,
SystemVPC: systemvpc.Deps{
VPC: d.vpcService,
@@ -54,12 +54,12 @@ func (d *Daemon) buildBedrockLaunchDeps() handlers_bedrock.LaunchDeps {
// interface) when no GPU manager is present: assigning a typed nil
// *daemonGPUSnapshotter instead would produce a non-nil interface wrapping a
// nil pointer, defeating admitCapacity's own "snapshotter == nil" check.
func (d *Daemon) buildBedrockServiceDeps() handlers_bedrock.ServiceDeps {
func (d *Daemon) buildBedrockServiceDeps() ochre.ServiceDeps {
clusterSize := 1
if d.clusterConfig != nil {
clusterSize = len(d.clusterConfig.Nodes)
}
deps := handlers_bedrock.ServiceDeps{
deps := ochre.ServiceDeps{
Config: d.config,
Launch: d.buildBedrockLaunchDeps(),
NodeID: d.node,
@@ -71,7 +71,7 @@ func (d *Daemon) buildBedrockServiceDeps() handlers_bedrock.ServiceDeps {
return deps
}
// daemonGPUSnapshotter adapts *gpu.Manager to handlers_bedrock's Snapshot-only
// daemonGPUSnapshotter adapts *gpu.Manager to domains/ochre's Snapshot-only
// capacity-check surface, keeping the gpu package out of that handler package
// the same way daemonGPUClaimer keeps it out of handlers_ec2_instance.
//
+9 -9
View File
@@ -44,11 +44,11 @@ import (
"github.com/mulgadc/spinifex/spinifex/domains/dns"
"github.com/mulgadc/spinifex/spinifex/domains/ec2/instancetypes"
handlers_ecr "github.com/mulgadc/spinifex/spinifex/domains/ecr"
"github.com/mulgadc/spinifex/spinifex/domains/ochre"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/foundation/state/clustersize"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/telemetry"
handlers_bedrock "github.com/mulgadc/spinifex/spinifex/handlers/bedrock"
handlers_ec2_account "github.com/mulgadc/spinifex/spinifex/handlers/ec2/account"
handlers_ec2_eigw "github.com/mulgadc/spinifex/spinifex/handlers/ec2/eigw"
handlers_ec2_eip "github.com/mulgadc/spinifex/spinifex/handlers/ec2/eip"
@@ -169,8 +169,8 @@ type Daemon struct {
ecsScheduler *handlers_ecs.Scheduler
rdsService *handlers_rds.Service
rdsReconciler *handlers_rds.Reconciler
bedrockService *handlers_bedrock.Service
bedrockReaper *handlers_bedrock.Reaper
bedrockService *ochre.Service
bedrockReaper *ochre.Reaper
acmService *acmdomain.ACMServiceImpl
acmRenewalWorker *acmdomain.Worker
ochreVectorService handlers_ochrevector.VectorService
@@ -1208,10 +1208,10 @@ func (d *Daemon) subscribeAll() error {
// touching JetStream directly).
if d.bedrockService != nil {
subs = append(subs,
natsSub{handlers_bedrock.SubjectEnsureEndpoint, handleNATSRequest(d.node, d.bedrockService.Ensure), "spinifex-workers"},
natsSub{handlers_bedrock.SubjectDescribeEndpoint, handleNATSRequest(d.node, d.bedrockService.Describe), "spinifex-workers"},
natsSub{handlers_bedrock.SubjectListEndpoints, handleNATSRequest(d.node, d.bedrockService.List), "spinifex-workers"},
natsSub{handlers_bedrock.SubjectDeleteEndpoint, handleNATSRequest(d.node, d.bedrockService.Delete), "spinifex-workers"},
natsSub{ochre.SubjectEnsureEndpoint, handleNATSRequest(d.node, d.bedrockService.Ensure), "spinifex-workers"},
natsSub{ochre.SubjectDescribeEndpoint, handleNATSRequest(d.node, d.bedrockService.Describe), "spinifex-workers"},
natsSub{ochre.SubjectListEndpoints, handleNATSRequest(d.node, d.bedrockService.List), "spinifex-workers"},
natsSub{ochre.SubjectDeleteEndpoint, handleNATSRequest(d.node, d.bedrockService.Delete), "spinifex-workers"},
)
}
@@ -2080,12 +2080,12 @@ func (d *Daemon) startCluster() error {
// Bedrock serving-endpoint lifecycle: the request-driven
// ensure/describe/list/delete surface, constructed synchronously since it
// touches no JetStream KV until its first request.
d.bedrockService = handlers_bedrock.NewService(d.natsConn, d.buildBedrockServiceDeps())
d.bedrockService = ochre.NewService(d.natsConn, d.buildBedrockServiceDeps())
// One leader across the cluster scrapes each serving endpoint's vLLM metrics
// and hands an idle GPU back; every node keeps serving the API. Without it a
// launched model owns its device until an operator deletes the endpoint.
d.bedrockReaper = handlers_bedrock.NewReaper(d.bedrockService, d.node, handlers_bedrock.ReaperDeps{})
d.bedrockReaper = ochre.NewReaper(d.bedrockService, d.node, ochre.ReaperDeps{})
d.shutdownWg.Go(func() {
defer func() {
if r := recover(); r != nil {
+2 -2
View File
@@ -10,9 +10,9 @@ import (
"time"
"github.com/mulgadc/bluebottle/pkg/masterkey"
"github.com/mulgadc/spinifex/spinifex/domains/ochre"
"github.com/mulgadc/spinifex/spinifex/foundation/telemetry"
gateway_bedrock "github.com/mulgadc/spinifex/spinifex/gateway/bedrock"
handlers_bedrock "github.com/mulgadc/spinifex/spinifex/handlers/bedrock"
handlers_ochrevector "github.com/mulgadc/spinifex/spinifex/handlers/ochrevector"
"github.com/mulgadc/spinifex/spinifex/network/host"
"github.com/mulgadc/spinifex/spinifex/providers/objectstore"
@@ -117,7 +117,7 @@ func (d *Daemon) startOchreVector() {
// Embedder/reranker are bundle members: DynamicEndpointResolver resolves
// each model id to the co-resident bundle VM's own port, in-process
// against d.bedrockService rather than a NATS round trip to itself.
endpointResolver := handlers_bedrock.NewDynamicEndpointResolver(d.bedrockService, nil, 0)
endpointResolver := ochre.NewDynamicEndpointResolver(d.bedrockService, nil, 0)
embedder := gateway_bedrock.NewEmbedder(endpointResolver)
// RerankModel unset leaves reranker nil: Query degrades to plain KNN
@@ -1,4 +1,4 @@
package handlers_bedrock
package ochre
import (
"context"
@@ -1,4 +1,4 @@
package handlers_bedrock
package ochre
import (
"context"
@@ -1,4 +1,4 @@
package handlers_bedrock
package ochre
import (
"fmt"
@@ -1,4 +1,4 @@
package handlers_bedrock
package ochre
import (
"testing"
@@ -1,4 +1,4 @@
package handlers_bedrock
package ochre
import (
awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers"
@@ -1,4 +1,4 @@
package handlers_bedrock
package ochre
import (
"context"
@@ -1,4 +1,4 @@
package handlers_bedrock
package ochre
import (
"context"
@@ -1,4 +1,4 @@
package handlers_bedrock
package ochre
import (
"errors"
@@ -1,4 +1,4 @@
package handlers_bedrock
package ochre
import (
"context"
@@ -1,4 +1,4 @@
package handlers_bedrock
package ochre
import (
"context"
@@ -1,4 +1,4 @@
package handlers_bedrock
package ochre
import (
"bufio"
@@ -1,4 +1,4 @@
package handlers_bedrock
package ochre
import (
"net/http"
@@ -1,4 +1,4 @@
package handlers_bedrock
package ochre
import (
"context"
@@ -9,7 +9,7 @@ import (
// ProvisionedEndpointAdapter adapts EndpointService to gateway_bedrock's
// EndpointProvisioner, the narrow surface the provisioned-throughput ops
// need. It exists because gateway_bedrock cannot import this package back
// (handlers_bedrock already imports gateway_bedrock for LookupServingSpec),
// (domains/ochre already imports gateway_bedrock for LookupServingSpec),
// so EndpointProvisioner is declared there with primitive-typed methods and
// this adapter is what actually calls EndpointService underneath.
type ProvisionedEndpointAdapter struct {
@@ -1,4 +1,4 @@
package handlers_bedrock
package ochre
import (
"context"
@@ -1,4 +1,4 @@
package handlers_bedrock
package ochre
import (
"context"
@@ -1,4 +1,4 @@
package handlers_bedrock
package ochre
import (
"context"
@@ -1,4 +1,4 @@
package handlers_bedrock
package ochre
import (
"context"
@@ -1,4 +1,4 @@
package handlers_bedrock
package ochre
import (
"context"
@@ -1,4 +1,4 @@
package handlers_bedrock
package ochre
import (
"context"
@@ -37,7 +37,7 @@ const livenessProbeTimeout = 500 * time.Millisecond
// DynamicEndpointResolver resolves a self-host model's base URL through the
// daemon's endpoint registry, and asks for a launch when there is nothing to
// resolve. It lives here rather than in gateway_bedrock because that package
// cannot import this one: handlers_bedrock already imports it for
// cannot import this one: domains/ochre already imports it for
// LookupServingSpec.
type DynamicEndpointResolver struct {
svc EndpointService
@@ -1,4 +1,4 @@
package handlers_bedrock
package ochre
import (
"context"
@@ -1,4 +1,4 @@
package handlers_bedrock
package ochre
import (
"context"
@@ -1,4 +1,4 @@
package handlers_bedrock
package ochre
import (
"context"
@@ -16,7 +16,7 @@ const defaultNATSTimeout = 30 * time.Second
// NATSEndpointService is the NATS-forwarding EndpointService client: the
// gateway process (or any other caller off the daemon's own node) uses this
// instead of importing handlers_bedrock's JetStream/launch dependencies.
// instead of importing domains/ochre's JetStream/launch dependencies.
type NATSEndpointService struct {
nc *nats.Conn
timeout time.Duration
@@ -1,4 +1,4 @@
package handlers_bedrock
package ochre
import (
"context"
@@ -1,4 +1,4 @@
package handlers_bedrock
package ochre
import (
awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers"
@@ -1,8 +1,8 @@
// Package handlers_bedrock owns placement and lifecycle of self-hosted vLLM
// Package ochre owns placement and lifecycle of self-hosted vLLM
// serving VMs: it decides where a model's endpoint runs, launches and probes
// the VM, and persists the result as the single source of truth other
// components (the gateway's inference path) read over NATS.
package handlers_bedrock
package ochre
import (
"errors"
@@ -1,4 +1,4 @@
package handlers_bedrock
package ochre
import (
"testing"
@@ -1,4 +1,4 @@
package handlers_bedrock
package ochre
import (
"context"
@@ -1,4 +1,4 @@
package handlers_bedrock
package ochre
import (
"testing"
@@ -1,4 +1,4 @@
package handlers_bedrock
package ochre
// NATS subjects the daemon subscribes and NATSEndpointService forwards to,
// following the "eks.MethodName" naming EKS uses for its own internal
@@ -1,4 +1,4 @@
package handlers_bedrock
package ochre
import (
"fmt"
@@ -1,4 +1,4 @@
package handlers_bedrock
package ochre
import (
"context"
+3 -3
View File
@@ -10,7 +10,7 @@ import (
// embedderHealthPath is the readiness route embeddingsProvider's warm-up
// probe polls -- TEI has no other health route, mirroring
// handlers/bedrock/readiness.go's default (non-vLLM) family selection.
// domains/ochre/readiness.go's default (non-vLLM) family selection.
const embedderHealthPath = "/health"
// embedderWarmupPollInterval spaces the background warm-up probe's checks,
@@ -91,8 +91,8 @@ func (p *warmupProbe) Ready() bool {
// embedderHealthPath and reports whether it returned HTTP 200. Any
// transport error (connection refused while TEI is still starting) or
// non-200 status counts as not-yet-ready, mirroring
// handlers/bedrock/readiness.go's probeOnce -- duplicated rather than
// imported, since handlers_bedrock already imports gateway_bedrock and Go
// domains/ochre/readiness.go's probeOnce -- duplicated rather than
// imported, since domains/ochre already imports gateway_bedrock and Go
// forbids the reverse.
func probeEmbedderHealth(client *http.Client, baseURL string) bool {
ctx, cancel := context.WithTimeout(context.Background(), embedderWarmupProbeTimeout)
+7 -7
View File
@@ -15,19 +15,19 @@ import (
)
// EndpointProvisioner is the narrow surface the provisioned-throughput ops
// need from the daemon's endpoint lifecycle (handlers_bedrock.EndpointService):
// need from the daemon's endpoint lifecycle (ochre.EndpointService):
// launch a pinned endpoint, read its state, and tear it down. It is declared
// here with primitive-typed methods rather than that package's input/output
// structs because handlers_bedrock already imports this package
// structs because domains/ochre already imports this package
// (LookupServingSpec), so importing it back here would cycle. See
// handlers_bedrock.ProvisionedEndpointAdapter for the real implementation
// ochre.ProvisionedEndpointAdapter for the real implementation
// over EndpointService; tests in this package use a stub instead.
type EndpointProvisioner interface {
// EnsurePinned requests a pinned, account-scoped endpoint for modelID.
EnsurePinned(ctx context.Context, accountID, modelID string) error
// EndpointState reports (accountID, modelID)'s current endpoint state:
// one of the endpointState* constants below, mirroring
// handlers_bedrock.EndpointState's string values.
// ochre.EndpointState's string values.
EndpointState(ctx context.Context, accountID, modelID string) (state string, err error)
// DeletePinned tears down (accountID, modelID)'s pinned endpoint.
// Idempotent: an already-absent endpoint is a success.
@@ -35,7 +35,7 @@ type EndpointProvisioner interface {
}
// Endpoint state strings EndpointProvisioner.EndpointState returns, mirroring
// handlers_bedrock.EndpointState's string values without importing that
// ochre.EndpointState's string values without importing that
// package back.
const (
endpointStateStarting = "STARTING"
@@ -51,7 +51,7 @@ const bedrockProvisionedBucket = "bedrock-provisioned"
const bedrockProvisionedHistory = 1
// ProvisionedModelRecord is the gateway control-plane state for one
// commitment. The VM it pins is daemon-owned (handlers_bedrock.EndpointRecord);
// commitment. The VM it pins is daemon-owned (ochre.EndpointRecord);
// this record is the AWS-shaped metadata layered on top of it.
type ProvisionedModelRecord struct {
ARN string `json:"arn"`
@@ -355,7 +355,7 @@ func UpdateProvisionedModelThroughput(ctx context.Context, accountID string, sto
// DeleteProvisionedModelThroughput tears down the pinned endpoint for
// input.ProvisionedModelId's model, then removes the record. An already-absent
// commitment is a no-op success, matching handlers_bedrock.Service.Delete's
// commitment is a no-op success, matching ochre.Service.Delete's
// own idempotence.
func DeleteProvisionedModelThroughput(ctx context.Context, accountID string, store *ProvisionedStore, input *bedrock.DeleteProvisionedModelThroughputInput) (*bedrock.DeleteProvisionedModelThroughputOutput, error) {
if input == nil || aws.StringValue(input.ProvisionedModelId) == "" {
+2 -2
View File
@@ -25,7 +25,7 @@ type stubEndpointProvisioner struct {
// state maps "accountID/modelID" to the endpoint state EndpointState
// reports; an absent entry reports "" (ABSENT), matching
// handlers_bedrock.StateAbsent's zero-value shape.
// ochre.StateAbsent's zero-value shape.
state map[string]string
ensureErr error
@@ -331,7 +331,7 @@ func TestDeleteProvisionedModelThroughput_RemovesEndpointAndRecord(t *testing.T)
}
// TestDeleteProvisionedModelThroughput_AbsentIsNoop mirrors
// handlers_bedrock.Service.Delete's own idempotence: deleting an
// ochre.Service.Delete's own idempotence: deleting an
// already-absent commitment must succeed, not error.
func TestDeleteProvisionedModelThroughput_AbsentIsNoop(t *testing.T) {
stub := newStubEndpointProvisioner()
+1 -1
View File
@@ -71,7 +71,7 @@ func daemonPortDescription(nodeID string) string {
}
// ensureDaemonENI describe-or-creates this node's ENI in subnetID. Mirrors
// handlers/bedrock's helper of the same shape: description-filtered
// domains/ochre's helper of the same shape: description-filtered
// describe first, create only on a miss, so a restart reuses the same
// address rather than leaking one per launch.
func ensureDaemonENI(ctx context.Context, vpcSvc vpcProvisioner, subnetID, nodeID string, groupIDs []string) (*daemonENI, error) {
+6 -6
View File
@@ -28,13 +28,13 @@ import (
ecrauth "github.com/mulgadc/spinifex/spinifex/domains/ecr/auth"
awsapi "github.com/mulgadc/spinifex/spinifex/domains/ecr/awsapi"
ecrregistry "github.com/mulgadc/spinifex/spinifex/domains/ecr/registry"
"github.com/mulgadc/spinifex/spinifex/domains/ochre"
"github.com/mulgadc/spinifex/spinifex/foundation/lifecycle/reconciler"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/gateway"
gateway_bedrock "github.com/mulgadc/spinifex/spinifex/gateway/bedrock"
gateway_ec2_instance "github.com/mulgadc/spinifex/spinifex/gateway/ec2/instance"
handlers_bedrock "github.com/mulgadc/spinifex/spinifex/handlers/bedrock"
handlers_ec2_instance "github.com/mulgadc/spinifex/spinifex/handlers/ec2/instance"
handlers_iam "github.com/mulgadc/spinifex/spinifex/handlers/iam"
handlers_ochrevector "github.com/mulgadc/spinifex/spinifex/handlers/ochrevector"
@@ -363,10 +363,10 @@ func launchService(config *config.ClusterConfig) error {
// returning ModelNotReadyException at once. Unset (the default) keeps the
// fail-fast contract: cold start is minutes, which no client retry spans.
bedrockEndpoints := parseBedrockEndpoints(os.Getenv("OCHRE_VLLM_ENDPOINTS"))
bedrockEndpointSvc := handlers_bedrock.NewNATSEndpointService(natsConn)
bedrockEndpointResolver := handlers_bedrock.NewDynamicEndpointResolver(
bedrockEndpointSvc := ochre.NewNATSEndpointService(natsConn)
bedrockEndpointResolver := ochre.NewDynamicEndpointResolver(
bedrockEndpointSvc, bedrockEndpoints, 0,
handlers_bedrock.WithColdStartWait(parseColdStartWait(os.Getenv("OCHRE_COLD_START_WAIT"))))
ochre.WithColdStartWait(parseColdStartWait(os.Getenv("OCHRE_COLD_START_WAIT"))))
// Guardrail topicPolicy's semantic match reuses this same endpoint
// resolver (the one every self-hosted model, including the embedding
@@ -390,10 +390,10 @@ func launchService(config *config.ClusterConfig) error {
// endpoint it commits to is requested through the same NATS endpoint
// service the dynamic resolver above uses, via an adapter satisfying
// gateway_bedrock's narrow EndpointProvisioner (see provisioned_adapter.go
// for why the adapter, not handlers_bedrock.EndpointService, is what
// for why the adapter, not ochre.EndpointService, is what
// gateway_bedrock depends on).
bedrockProvisioned := gateway_bedrock.NewProvisionedStore(js, nodeConfig.Region,
handlers_bedrock.NewProvisionedEndpointAdapter(bedrockEndpointSvc))
ochre.NewProvisionedEndpointAdapter(bedrockEndpointSvc))
// Bedrock guardrails: control-plane CRUD only at this stage — the record
// stores the full policy config so a later stage's filter engine and