Skip to content

Maintain connections from the management-controller to all "manage" access points #101

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
ted-ross wants to merge 6 commits into
base: main
Choose a base branch
from
Open

Maintain connections from the management-controller to all "manage" access points #101

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
a0156b6
Syntax cleanup and added a sequence number to heartbeat messages to d…
ted-ross May 6, 2026
bbde10c
Disambiguated the global 'namespace' from argument names.
ted-ross May 6, 2026
da7f5a7
Fixed incorrect reference to the standalone namespace for non-standal…
ted-ross May 6, 2026
910ccca
Added notification of site deletion so sync heartbeats are not sent t…
ted-ross May 6, 2026
8fe0237
Minor whitespace cleanup.
ted-ross May 6, 2026
07e02c2
Fixes #92 - Syntax cleanup. Updated the algorithm to maintain connec…
ted-ross May 6, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 7 additions & 2 deletions components/management-controller/src/api-admin.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,7 @@

import { IncomingForm } from 'formidable';
import { ClientFromPool, queryWithContext } from './db.js';
import { SiteIngressChanged, LinkChanged } from './sync-management.js';
import { SiteIngressChanged, LinkChanged, SiteDeleted } from './sync-management.js';
import { Log } from '@skupperx/modules/log'
import { ManageIngressAdded, LinkAddedOrDeleted, ManageIngressDeleted } from './site-deployment-state.js';
import { ValidateAndNormalizeFields, IsValidUuid, UniquifyName } from '@skupperx/modules/util';
Expand Down Expand Up @@ -547,7 +547,12 @@ const deleteBackboneSite = async function(req, res) {
await client.query("DELETE FROM TlsCertificates WHERE Id = $1", [row.certificate])
}
}
})
});

//
// Notify the state-sync module that the site is no longer here.
//
SiteDeleted(sid);

res.status(returnStatus).end();
await WatchNotify('InteriorSites', sid);
Expand Down
57 changes: 26 additions & 31 deletions components/management-controller/src/backbone-links.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,19 +32,19 @@ let controller_name;
let tls_ca;
let tls_cert;
let tls_key;
let bbConnections = {};
let manageConnections = {};
let registrations = [];

const createConnection = async function(bbid, row) {
bbConnections[bbid] = {
async function createConnection(apid, row) {
manageConnections[apid] = {
toDelete: false,
host: row.hostname,
port: row.port,
};

Log(`Connecting to Access Point: ${row.hostname}:${row.port}`);
bbConnections[bbid].conn = OpenConnection(
`Backbone-management-${bbid}`,
manageConnections[apid].conn = OpenConnection(
`Backbone-management-${apid}`,
row.hostname,
row.port,
'tls',
Expand All @@ -53,53 +53,48 @@ const createConnection = async function(bbid, row) {
tls_key);

for (const reg of registrations) {
await reg.onLinkAdded(bbid, bbConnections[bbid].conn);
await reg.onLinkAdded(apid, manageConnections[apid].conn);
}
}

const deleteConnection = async function(bbid) {
let conn = bbConnections[bbid].conn;
async function deleteConnection(apid) {
let conn = manageConnections[apid].conn;
CloseConnection(conn);
delete bbConnections[bbid];
delete manageConnections[apid];

for (const reg of registrations) {
await reg.onLinkDeleted(bbid);
await reg.onLinkDeleted(apid);
}
}

const reconcileBackboneConnections = async function() {
async function reconcileBackboneConnections() {
let reschedule_delay = 30000;
const client = await ClientFromPool('system');
try {
await client.query('BEGIN');
const result = await client.query(
"SELECT BackboneAccessPoints.*, InteriorSites.Backbone " +
"FROM BackboneAccessPoints " +
"JOIN InteriorSites ON InteriorSites.Id = InteriorSite " +
"JOIN Backbones ON Backbones.Id = InteriorSites.Backbone " +
"WHERE BackboneAccessPoints.Lifecycle = 'ready' and Kind = 'manage'");
const result = await client.query("SELECT * FROM BackboneAccessPoints WHERE Lifecycle = 'ready' and Kind = 'manage'");
let db_rows = {};
for (const row of result.rows) {
if (!db_rows[row.backbone]) {
db_rows[row.backbone] = row;
}
}

for (const bbid of Object.keys(bbConnections)) {
bbConnections[bbid].toDelete = true;
for (const apid of Object.keys(manageConnections)) {
manageConnections[apid].toDelete = true;
}

for (const [bbid, row] of Object.entries(db_rows)) {
if (bbConnections[bbid]) {
bbConnections[bbid].toDelete = false;
for (const row of result.rows) {
if (manageConnections[row.id]) {
manageConnections[row.id].toDelete = false;
} else {
await createConnection(bbid, row);
await createConnection(row.id, row);
}
}

for (const bbid of Object.keys(bbConnections)) {
if (bbConnections[bbid].toDelete) {
await deleteConnection(bbid);
for (const apid of Object.keys(manageConnections)) {
if (manageConnections[apid].toDelete) {
await deleteConnection(apid);
}
}

Expand All @@ -114,7 +109,7 @@ const reconcileBackboneConnections = async function() {
}
}

const resolveTLSData = async function() {
async function resolveTLSData() {
let reschedule_delay = 1000;
const client = await ClientFromPool('system');
try {
Expand All @@ -139,13 +134,13 @@ const resolveTLSData = async function() {
}

if (count != 3) {
throw(Error(`Unexpected set of values from TLS secret data - expected 3, got ${count}`));
throw new Error(`Unexpected set of values from TLS secret data - expected 3, got ${count}`);
}

reschedule_delay = -1;
setTimeout(reconcileBackboneConnections, 0);
} else {
throw(Error(`Expected to find a TlsCertificate record for ready controller: ${result.rows[0].certificate}`));
throw new Error(`Expected to find a TlsCertificate record for ready controller: ${result.rows[0].certificate}`);
}
}
await client.query('COMMIT');
Expand All @@ -161,7 +156,7 @@ const resolveTLSData = async function() {
}
}

const resolveControllerRecord = async function() {
async function resolveControllerRecord() {
let reschedule_delay = -1;
const client = await ClientFromPool('system');
try {
Expand All @@ -188,7 +183,7 @@ const resolveControllerRecord = async function() {
}

export async function RegisterHandler(onAdded, onDeleted) {
for (const [key, value] of Object.entries(bbConnections)) {
for (const [key, value] of Object.entries(manageConnections)) {
await onAdded(key, value.conn);
}

Expand Down
2 changes: 1 addition & 1 deletion components/management-controller/src/mc-main.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@ import yaml from 'yaml';
import fs from 'node:fs';
import rhea from 'rhea';
import * as bbLinks from './backbone-links.js';
import * as externalVans from './external-vans.js';
import * as externalVans from './external-vans.js';
import * as certs from './certs.js';
import * as prune from './prune.js';
import * as db from './db.js';
Expand Down
52 changes: 28 additions & 24 deletions components/management-controller/src/sync-management.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,7 @@ import { Log } from '@skupperx/modules/log'
import { API_CONTROLLER_ADDRESS } from '@skupperx/modules/common'
import { ClientFromPool } from './db.js';
import { LoadSecret } from '@skupperx/modules/kube'
import { CLASS_MEMBER, CLASS_BACKBONE, AddConnection, DeleteConnection, UpdateLocalState, Start as StateSyncStart, CLASS_MANAGEMENT } from '@skupperx/modules/state-sync'
import { CLASS_MEMBER, CLASS_BACKBONE, AddConnection, DeleteConnection, UpdateLocalState, Start as StateSyncStart, CLASS_MANAGEMENT, DeletePeer } from '@skupperx/modules/state-sync'
import { onMewMember, StateRequest } from './sync-application.js';
import { RegisterHandler } from './backbone-links.js';
import { HashOfSecret, HashOfData } from './resource-templates.js';
Expand Down Expand Up @@ -78,7 +78,7 @@ export async function GetBackboneAccessPoints_TX(client, siteId, initialOnly = f
//=========================================================================================================================
// Backbone Site Handlers
//=========================================================================================================================
const onNewBackboneSite = async function(peerId) {
async function onNewBackboneSite(peerId) {
//
// peerId identifies the row in InteriorSites
//
Expand Down Expand Up @@ -174,11 +174,11 @@ const onNewBackboneSite = async function(peerId) {
return [localState, remoteState];
}

const onLostBackbone = async function(peerId) {
async function onLostBackbone(peerId) {
// Nothing to do here - Consider adding status to the schema to indicate a stale site
}

const onStateChangeBackbone = async function(peerId, stateKey, hash, data) {
async function onStateChangeBackbone(peerId, stateKey, hash, data) {
//
// Notes:
// This will update the access point with host/port on initial site creation.
Expand Down Expand Up @@ -215,7 +215,7 @@ const onStateChangeBackbone = async function(peerId, stateKey, hash, data) {
}
}

const getStateTlsBackboneSite = async function(siteId) {
async function getStateTlsBackboneSite(siteId) {
var hash = null;
var data = null;
const client = await ClientFromPool('system');
Expand All @@ -240,7 +240,7 @@ const getStateTlsBackboneSite = async function(siteId) {
return [hash, data];
}

const getStateTlsMemberSite = async function(siteId) {
async function getStateTlsMemberSite(siteId) {
var hash = null;
var data = null;
const client = await ClientFromPool('system');
Expand All @@ -265,7 +265,7 @@ const getStateTlsMemberSite = async function(siteId) {
return [hash, data];
}

const getStateTlsServer = async function(apid) {
async function getStateTlsServer(apid) {
var hash = null;
var data = null;
const client = await ClientFromPool('system');
Expand All @@ -290,7 +290,7 @@ const getStateTlsServer = async function(apid) {
return [hash, data];
}

const getStateAccessPoint = async function(apId) {
async function getStateAccessPoint(apId) {
var hash = null;
var data = null;
const client = await ClientFromPool('system');
Expand Down Expand Up @@ -318,7 +318,7 @@ const getStateAccessPoint = async function(apId) {
return [hash, data];
}

const getStateBackboneLink = async function(linkId) {
async function getStateBackboneLink(linkId) {
var hash = null;
var data = null;
const client = await ClientFromPool('system');
Expand Down Expand Up @@ -347,7 +347,7 @@ const getStateBackboneLink = async function(linkId) {
return [hash, data];
}

const getStateMemberLink = async function(linkId) {
async function getStateMemberLink(linkId) {
var hash = null;
var data = null;
const client = await ClientFromPool('system');
Expand Down Expand Up @@ -376,7 +376,7 @@ const getStateMemberLink = async function(linkId) {
return [hash, data];
}

const onStateRequestBackbone = async function(peerId, stateKey) { // => [hash, data]
async function onStateRequestBackbone(peerId, stateKey) {
var hash = null;
var data = null;

Expand All @@ -398,7 +398,7 @@ const onStateRequestBackbone = async function(peerId, stateKey) { // => [hash, d
//=========================================================================================================================
// Member Site Handlers
//=========================================================================================================================
const onNewMember = async function(peerId) {
async function onNewMember(peerId) {
//
// peerId identifies the row in MemberSites
//
Expand Down Expand Up @@ -470,15 +470,15 @@ const onNewMember = async function(peerId) {
return [localState, remoteState];
}

const onLostMember = async function(peerId) {
async function onLostMember(peerId) {
// TODO
}

const onStateChangeMember = async function(peerId, stateKey, hash, data) {
async function onStateChangeMember(peerId, stateKey, hash, data) {
// There is no local state on a member site
}

const onStateRequestMember = async function(peerId, stateKey) {
async function onStateRequestMember(peerId, stateKey) {
var hash = null;
var data = null;

Expand All @@ -497,7 +497,7 @@ const onStateRequestMember = async function(peerId, stateKey) {
//=========================================================================================================================
// Sync Handlers
//=========================================================================================================================
const onNewPeer = async function(peerId, peerClass) {
async function onNewPeer(peerId, peerClass) {
var localState;
var remoteState;
peers[peerId] = {
Expand All @@ -513,7 +513,7 @@ const onNewPeer = async function(peerId, peerClass) {
return [localState, remoteState];
}

const onPeerLost = async function(peerId) {
async function onPeerLost(peerId) {
const peer = peers[peerId];
if (!!peer) {
if (peer.pClass == CLASS_MEMBER) {
Expand All @@ -526,7 +526,7 @@ const onPeerLost = async function(peerId) {
}
}

const onStateChange = async function(peerId, stateKey, hash, data) {
async function onStateChange(peerId, stateKey, hash, data) {
const peer = peers[peerId];
if (!!peer) {
if (peer.pClass == CLASS_MEMBER) {
Expand All @@ -537,7 +537,7 @@ const onStateChange = async function(peerId, stateKey, hash, data) {
}
}

const onStateRequest = async function(peerId, stateKey) {
async function onStateRequest(peerId, stateKey) {
var hash = null;
var data = null;
const peer = peers[peerId];
Expand All @@ -551,7 +551,7 @@ const onStateRequest = async function(peerId, stateKey) {
return [hash, data];
}

const onPing = async function(peerId) {
async function onPing(peerId) {
const client = await ClientFromPool('system');
try {
await client.query("BEGIN");
Expand Down Expand Up @@ -580,12 +580,12 @@ const onPing = async function(peerId) {
//=========================================================================================================================
// Backbone Link Handlers
//=========================================================================================================================
const onLinkAdded = async function(backboneId, conn) {
await AddConnection(backboneId, conn);
async function onLinkAdded(key, conn) {
await AddConnection(key, conn);
}

const onLinkDeleted = async function(backboneId) {
await DeleteConnection(backboneId);
async function onLinkDeleted(key) {
await DeleteConnection(key);
}

//=========================================================================================================================
Expand Down Expand Up @@ -732,6 +732,10 @@ export async function NewIngressAvailable(siteId) {
}
}

export async function SiteDeleted(siteId) {
DeletePeer(siteId);
}

export async function Start() {
await StateSyncStart(CLASS_MANAGEMENT, 'mc', API_CONTROLLER_ADDRESS, onNewPeer, onPeerLost, onStateChange, onStateRequest, onPing);
await RegisterHandler(onLinkAdded, onLinkDeleted);
Expand Down
6 changes: 2 additions & 4 deletions components/site-controller/src/sc-main.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -73,16 +73,14 @@ export async function Main() {

Log(`Site-Id : ${site_id}`);
let conn;
if ( PLATFORM == 'sk2' ) {
if (PLATFORM == 'sk2') {
Log('Waiting for skupper-router pod to be Running...');
if (!kube.waitPodsRunning(STANDALONE_NAMESPACE, 'application=skupper-router')) {
if (!kube.waitPodsRunning(kube.Namespace(), 'application=skupper-router')) {
Log('Skupper-router is not running, exiting');
process.exit(1);
}
let certs = await GetLocalRouterCerts();
conn = amqp.OpenConnection('LocalRouter', 'skupper-router-local', '5671', 'tls', certs.ca, certs.cert, certs.key);
} else {
conn = amqp.OpenConnection('LocalRouter');
}
await router.Start(conn);
if (PLATFORM != 'sk2') {
Expand Down
Loading