Method createSessionOnWorker

controlplane/session_mgr.go:540–633

(ctx context.Context, username string, pid int32, memoryLimit string, threads int, worker *ManagedWorker, protocol string, retireOnFailure bool, lease connectionLease)

			sm.log.Error("Failed to load user persistent secrets; session starts without them.",
				"pid", pid, "worker", worker.ID, "user", username, "error", secretErr)
		}
	}

	sessionToken, secretWarnings, err := worker.CreateSession(ctx, username, memoryLimit, threads, secretStatements, pid)
	if err != nil {
		sm.log.Warn("Failed to create session on worker.",
			"pid", pid,
			"worker", worker.ID,
			"user", username,
			"protocol", protocol,
			"duration", time.Since(createStart),
			"retire_on_failure", retireOnFailure,
			"error", err,
		)
		if retireOnFailure {
			sm.pool.RetireWorkerIfNoSessions(worker.ID)
		}
		return 0, nil, fmt.Errorf("create session on worker %d: %w", worker.ID, err)
	}

	for _, w := range secretWarnings {
		sm.log.Warn("User persistent secret replay warning.",
			"pid", pid, "worker", worker.ID, "user", username, "warning", w)
	}

	executor := flightclient.NewFlightExecutorFromClientWithQueryLogLimiter(
		worker.client,
		sessionToken,
		worker.workerQueryLogLimiter(),
	)
	executor.SetControlMetadata(worker.ID, worker.OwnerCPInstanceID(), worker.OwnerEpoch())

	if pid == 0 {
		pid = reservePID(globalNextPID)
	}

	session := &ManagedSession{
		PID:          pid,
		Username:     username,
		WorkerID:     worker.ID,
		Protocol:     protocol,
		StartedAt:    time.Now().UTC(),
		SessionToken: sessionToken,
		Executor:     executor,
		lease:        lease,
	}

	sm.mu.Lock()
	if sm.lifecycle.isClosed() {
		sm.mu.Unlock()
		sm.cleanupUnregisteredWorkerSession(worker, session)
		return 0, nil, ErrSessionManagerDraining
	}
	sm.sessions[pid] = session
	sm.byWorker[worker.ID] = append(sm.byWorker[worker.ID], pid)
	sessionCount := len(sm.sessions)
	workerSessionCount := len(sm.byWorker[worker.ID])
	sm.mu.Unlock()