createOrgStack creates an isolated pool + session manager for an org.
(tc *configstore.OrgConfig)
| 97 | } |
| 98 | if _, err := tr.createOrgStack(tc); err != nil { |
| 99 | slog.Error("Failed to create org stack.", "org", tc.Name, "error", err) |
| 100 | continue |
| 101 | } |
| 102 | } |
| 103 | |
| 104 | return tr, nil |
| 105 | } |
| 106 | |
| 107 | // createOrgStack creates an isolated pool + session manager for an org. |
| 108 | func (tr *OrgRouter) createOrgStack(tc *configstore.OrgConfig) (*OrgStack, error) { |
| 109 | if tc == nil { |
| 110 | return nil, fmt.Errorf("org config is required") |
| 111 | } |
| 112 | finishCreation, err := tr.beginOrgStackCreation(tc.Name) |
| 113 | if err != nil { |
| 114 | return nil, err |
| 115 | } |
| 116 | defer finishCreation() |
| 117 | |
| 118 | // A config callback may have waited behind an older mutation. Resolve the |
| 119 | // authoritative config only after acquiring the per-org slot so stale |
| 120 | // callbacks cannot construct a generation the current snapshot rejects. |
| 121 | latest, state := tr.latestOrgStackState(tc.Name, &configstore.Snapshot{ |
| 122 | Orgs: map[string]*configstore.OrgConfig{tc.Name: tc}, |
| 123 | }) |
| 124 | if state != orgStackEnsurePresent { |
| 125 | return nil, fmt.Errorf("current config does not allow an org stack for org %s", tc.Name) |
| 126 | } |
| 127 | |
| 128 | return tr.createOrgStackWhileMutationHeld(latest) |
| 129 | } |
| 130 | |
| 131 | // createOrgStackWhileMutationHeld constructs and publishes one org generation. |
| 132 | // The caller must hold the org's mutation slot from before config selection |
| 133 | // through this function's return. |
| 134 | func (tr *OrgRouter) createOrgStackWhileMutationHeld(tc *configstore.OrgConfig) (*OrgStack, error) { |
| 135 | |
| 136 | ctx, cancel := context.WithCancel(context.Background()) |
| 137 | |
| 138 | // Per-org worker cap. 0 = unbounded (the cluster autoscaler / node capacity |
| 139 | // is the only ceiling). There is no global/cluster-default fallback. |
| 140 | maxWorkers := tc.MaxWorkers |
| 141 | |
| 142 | pool := NewOrgReservedPool(tr.sharedPool, tc.Name, maxWorkers, workerImageForOrg(tc, tr.baseCfg.WorkerImage), tr.stsBroker) |
| 143 | activator := NewSharedWorkerActivator(tr.sharedPool, tr.stsBroker, tr.globalCfg.DuckLakeDefaultSpecVersion, func(orgID string) (*configstore.OrgConfig, error) { |
| 144 | snap := tr.configStore.Snapshot() |
| 145 | if snap == nil { |
| 146 | return nil, fmt.Errorf("config snapshot unavailable for org %s", orgID) |
| 147 | } |
| 148 | org, ok := snap.Orgs[orgID] |
| 149 | if !ok { |
| 150 | return nil, fmt.Errorf("org %s not found in config snapshot", orgID) |
| 151 | } |
| 152 | return org, nil |
| 153 | }) |
| 154 | activator.resolveDucklingStatus = tr.resolveDucklingStatus |
| 155 | activator.setMigrating = tr.SetMigrating |
| 156 | activator.clearMigrating = tr.ClearMigrating |