YuzuZensai/Minikura
1
0
Fork 0
mirror of https://github.com/YuzuZensai/Minikura.git synced 2026-09-13 10:49:21 +00:00
Code Issues Packages Projects Releases Wiki Activity

♻️ refactor: migrate to Go Kubernetes operator

Browse Source
This commit is contained in:
yuzu
2026-08-13 02:19:27 +07:00
parent 826464a9e5
commit 91c07eb76a
50 changed files with 532 additions and 2930 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/k8s-operator/Dockerfile
-32
View File
@@ -1,32 +0,0 @@
FROM node:18-alpine AS build
WORKDIR /app
# Copy package files
COPY package.json ./
COPY tsconfig.json ./
# Copy source files
COPY src/ ./src/
# Install dependencies
RUN npm install
# Build
RUN npm run build
# Create production image
FROM node:18-alpine
WORKDIR /app
# Copy package.json and built files
COPY --from=build /app/package.json ./
COPY --from=build /app/dist ./dist
# Install production dependencies
RUN npm install --production
# Set environment variables
ENV NODE_ENV=production
# Run
CMD ["node", "dist/index.js"]
packages/k8s-operator/package.json
-33
View File
@@ -1,33 +0,0 @@
{
"name": "@minikura/k8s-operator",
"version": "1.0.0",
"description": "Kubernetes operator for Minikura that syncs database to Kubernetes resources",
"main": "dist/index.js",
"type": "module",
"scripts": {
"build": "tsc",
"start": "node dist/index.js",
"dev": "tsx watch src/index.ts",
"dev:bun": "bun --watch src/index.ts",
"watch": "tsx watch src/index.ts",
"apply-crds": "tsx src/scripts/apply-crds.ts",
"typecheck": "tsc --noEmit"
},
"dependencies": {
"@kubernetes/client-node": "^1.4.0",
"@minikura/api": "workspace:*",
"@minikura/db": "workspace:*",
"dotenv-mono": "^1.5.1",
"node-fetch": "^3.3.2",
"pg": "^8.23.0",
"pino": "^10.3.1",
"pino-pretty": "^13.1.3",
"undici": "^8.10.0",
"yaml": "^2.9.0"
},
"devDependencies": {
"@types/node": "^26.2.0",
"tsx": "^4.23.12",
"typescript": "^7.0.2"
}
}
packages/k8s-operator/src/config/constants.ts
-40
View File
@@ -1,40 +0,0 @@
import { dotenvLoad } from "dotenv-mono";
const _dotenv = dotenvLoad();
export { API_GROUP, LABEL_PREFIX } from "@minikura/api";
export const API_VERSION = "v1alpha1";
export const KUBERNETES_NAMESPACE_ENV = process.env.KUBERNETES_NAMESPACE;
export const NAMESPACE = process.env.KUBERNETES_NAMESPACE || "minikura";
export const ENABLE_CRD_REFLECTION = process.env.ENABLE_CRD_REFLECTION === "true";
export const RESOURCE_TYPES = {
MINECRAFT_SERVER: {
kind: "MinecraftServer",
plural: "minecraftservers",
singular: "minecraftserver",
shortNames: ["mcs"],
},
REVERSE_PROXY_SERVER: {
kind: "ReverseProxyServer",
plural: "reverseproxyservers",
singular: "reverseproxyserver",
shortNames: ["rps"],
},
};
export const SYNC_INTERVAL = 30 * 1000;
export const IMAGES = {
MINECRAFT: "itzg/minecraft-server",
REVERSE_PROXY: "itzg/minecraft-server",
};
export const DEFAULTS = {
MEMORY: "1G",
CPU_REQUEST: "250m",
CPU_LIMIT: "1000m",
STORAGE_SIZE: "1Gi",
};
packages/k8s-operator/src/config/resource-defaults.ts
-15
View File
@@ -1,15 +0,0 @@
export const RESOURCE_DEFAULTS = {
server: {
memory: "1G",
javaMemoryFactor: 0.8,
},
proxy: {
memory: "512M",
javaMemoryFactor: 0.8,
},
} as const;
export const JAVA_MEMORY_FACTOR = 0.8;
export const DEFAULT_SERVER_MEMORY = "1G";
export const DEFAULT_PROXY_MEMORY = "512M";
packages/k8s-operator/src/controllers/base-controller.ts
-53
View File
@@ -1,53 +0,0 @@
import type { PrismaClient } from "@minikura/db";
import type { Logger } from "pino";
import { SYNC_INTERVAL } from "../config/constants";
import { KubernetesClient } from "../utils/k8s-client";
import { createLogger } from "../utils/logger";
export abstract class BaseController {
protected prisma: PrismaClient;
protected k8sClient: KubernetesClient;
protected namespace: string;
protected logger: Logger;
private intervalId: ReturnType<typeof setInterval> | null = null;
constructor(prisma: PrismaClient, namespace: string) {
this.prisma = prisma;
this.k8sClient = KubernetesClient.getInstance();
this.namespace = namespace;
this.logger = createLogger({ controller: this.getControllerName() });
}
public startWatching(): void {
this.logger.info(
{ namespace: this.namespace, syncInterval: SYNC_INTERVAL },
"Starting controller watch loop"
);
this.syncResources().catch((err) => {
this.logger.error({ err }, "Error during initial resource synchronization");
});
this.intervalId = setInterval(() => {
this.syncResources().catch((err) => {
this.logger.error({ err }, "Error during periodic resource synchronization");
});
}, SYNC_INTERVAL);
this.logger.debug(
{ intervalMs: SYNC_INTERVAL },
"Polling interval established for resource synchronization"
);
}
public stopWatching(): void {
if (this.intervalId) {
clearInterval(this.intervalId);
this.intervalId = null;
this.logger.info("Controller watch loop stopped");
}
}
protected abstract getControllerName(): string;
protected abstract syncResources(): Promise<void>;
}
packages/k8s-operator/src/controllers/reverse-proxy-controller.ts
-124
View File
@@ -1,124 +0,0 @@
import type { CustomEnvironmentVariable, ReverseProxyServer } from "@minikura/db";
import {
createReverseProxyServer,
deleteReverseProxyServer,
} from "../resources/reverseProxyServer";
import type { ReverseProxyConfig } from "../types";
import { BaseController } from "./base-controller";
type ReverseProxyWithEnvVars = ReverseProxyServer & {
env_variables: CustomEnvironmentVariable[];
};
export class ReverseProxyController extends BaseController {
private deployedProxies = new Map<string, ReverseProxyWithEnvVars>();
protected getControllerName(): string {
return "ReverseProxyController";
}
protected async syncResources(): Promise<void> {
try {
const appsApi = this.k8sClient.getAppsApi();
const coreApi = this.k8sClient.getCoreApi();
const networkingApi = this.k8sClient.getNetworkingApi();
const proxies = (await this.prisma.reverseProxyServer.findMany({
include: {
env_variables: true,
},
})) as ReverseProxyWithEnvVars[];
const currentProxyIds = new Set(proxies.map((proxy) => proxy.id));
for (const [proxyId, proxy] of this.deployedProxies.entries()) {
if (!currentProxyIds.has(proxyId)) {
this.logger.info(
{ proxyId, proxyType: proxy.type },
"Reverse proxy removed from database, deleting K8s resources"
);
await deleteReverseProxyServer(proxy.id, proxy.type, appsApi, coreApi, this.namespace);
this.deployedProxies.delete(proxyId);
}
}
for (const proxy of proxies) {
const deployedProxy = this.deployedProxies.get(proxy.id);
if (!deployedProxy || this.hasProxyChanged(deployedProxy, proxy)) {
const action = !deployedProxy ? "Creating" : "Updating";
this.logger.info(
{
proxyId: proxy.id,
proxyType: proxy.type,
action: action.toLowerCase(),
externalAddress: proxy.external_address,
externalPort: proxy.external_port,
listenPort: proxy.listen_port,
},
`${action} reverse proxy server in Kubernetes`
);
const proxyConfig: ReverseProxyConfig = {
id: proxy.id,
external_address: proxy.external_address,
external_port: proxy.external_port,
listen_port: proxy.listen_port,
description: proxy.description,
apiKey: proxy.api_key,
type: proxy.type,
memory: proxy.memory,
service_type: proxy.service_type,
env_variables: proxy.env_variables?.map((ev) => ({
key: ev.key,
value: ev.value,
})),
};
await createReverseProxyServer(
proxyConfig,
appsApi,
coreApi,
networkingApi,
this.namespace
);
this.deployedProxies.set(proxy.id, { ...proxy });
}
}
} catch (error) {
this.logger.error({ err: error }, "Failed to sync reverse proxy servers to Kubernetes");
throw error;
}
}
private hasProxyChanged(
oldProxy: ReverseProxyWithEnvVars,
newProxy: ReverseProxyWithEnvVars
): boolean {
const basicPropsChanged =
oldProxy.external_address !== newProxy.external_address ||
oldProxy.external_port !== newProxy.external_port ||
oldProxy.listen_port !== newProxy.listen_port ||
oldProxy.description !== newProxy.description ||
oldProxy.service_type !== newProxy.service_type ||
oldProxy.memory !== newProxy.memory ||
oldProxy.type !== newProxy.type;
if (basicPropsChanged) return true;
const oldEnvVars = oldProxy.env_variables || [];
const newEnvVars = newProxy.env_variables || [];
if (oldEnvVars.length !== newEnvVars.length) return true;
for (const newEnv of newEnvVars) {
const oldEnv = oldEnvVars.find((e) => e.key === newEnv.key);
if (!oldEnv || oldEnv.value !== newEnv.value) {
return true;
}
}
return false;
}
}
packages/k8s-operator/src/controllers/server-controller.ts
-107
View File
@@ -1,107 +0,0 @@
import type { CustomEnvironmentVariable, Server } from "@minikura/db";
import { createServer, deleteServer } from "../resources/server";
import type { ServerConfig } from "../types";
import { BaseController } from "./base-controller";
type ServerWithEnvVars = Server & {
env_variables: CustomEnvironmentVariable[];
};
export class ServerController extends BaseController {
private deployedServers = new Map<string, ServerWithEnvVars>();
protected getControllerName(): string {
return "ServerController";
}
protected async syncResources(): Promise<void> {
try {
const appsApi = this.k8sClient.getAppsApi();
const coreApi = this.k8sClient.getCoreApi();
const networkingApi = this.k8sClient.getNetworkingApi();
const servers = (await this.prisma.server.findMany({
include: {
env_variables: true,
},
})) as ServerWithEnvVars[];
const currentServerIds = new Set(servers.map((server) => server.id));
for (const [serverId, server] of this.deployedServers.entries()) {
if (!currentServerIds.has(serverId)) {
this.logger.info(
{ serverId, serverName: server.id },
"Server removed from database, deleting K8s resources"
);
await deleteServer(serverId, appsApi, coreApi, this.namespace);
this.deployedServers.delete(serverId);
}
}
for (const server of servers) {
const deployedServer = this.deployedServers.get(server.id);
if (!deployedServer || this.hasServerChanged(deployedServer, server)) {
const action = !deployedServer ? "Creating" : "Updating";
this.logger.info(
{
serverId: server.id,
serverType: server.type,
action: action.toLowerCase(),
memory: server.memory,
port: server.listen_port,
},
`${action} Minecraft server in Kubernetes`
);
const serverConfig: ServerConfig = {
id: server.id,
type: server.type,
apiKey: server.api_key,
description: server.description,
listen_port: server.listen_port,
memory: server.memory,
service_type: server.service_type,
env_variables: server.env_variables?.map((ev) => ({
key: ev.key,
value: ev.value,
})),
};
await createServer(serverConfig, appsApi, coreApi, networkingApi, this.namespace);
this.deployedServers.set(server.id, { ...server });
}
}
} catch (error) {
this.logger.error({ err: error }, "Failed to sync servers to Kubernetes");
throw error;
}
}
private hasServerChanged(oldServer: ServerWithEnvVars, newServer: ServerWithEnvVars): boolean {
const basicPropsChanged =
oldServer.type !== newServer.type ||
oldServer.listen_port !== newServer.listen_port ||
oldServer.description !== newServer.description ||
oldServer.service_type !== newServer.service_type ||
oldServer.memory !== newServer.memory;
if (basicPropsChanged) return true;
const oldEnvVars = oldServer.env_variables || [];
const newEnvVars = newServer.env_variables || [];
if (oldEnvVars.length !== newEnvVars.length) return true;
for (const newEnv of newEnvVars) {
const oldEnv = oldEnvVars.find((e) => e.key === newEnv.key);
if (!oldEnv || oldEnv.value !== newEnv.value) {
return true;
}
}
return false;
}
}
packages/k8s-operator/src/crds/rbac.ts
-154
View File
@@ -1,154 +0,0 @@
import { API_GROUP, NAMESPACE } from "../config/constants";
export const minikuraNamespace = {
apiVersion: "v1",
kind: "Namespace",
metadata: {
name: NAMESPACE,
},
};
export const minikuraServiceAccount = {
apiVersion: "v1",
kind: "ServiceAccount",
metadata: {
name: "minikura-operator",
namespace: NAMESPACE,
},
};
export const minikuraClusterRole = {
apiVersion: "rbac.authorization.k8s.io/v1",
kind: "ClusterRole",
metadata: {
name: "minikura-operator-role",
},
rules: [
{
apiGroups: [""],
resources: ["configmaps", "services", "secrets"],
verbs: ["get", "list", "watch", "create", "update", "patch", "delete"],
},
{
apiGroups: ["apps"],
resources: ["deployments", "statefulsets"],
verbs: ["get", "list", "watch", "create", "update", "patch", "delete"],
},
{
apiGroups: ["networking.k8s.io"],
resources: ["ingresses"],
verbs: ["get", "list", "watch", "create", "update", "patch", "delete"],
},
{
apiGroups: ["apiextensions.k8s.io"],
resources: ["customresourcedefinitions"],
verbs: ["get", "list", "watch", "create", "update", "patch", "delete"],
},
{
apiGroups: [API_GROUP],
resources: ["minecraftservers", "velocityproxies"],
verbs: ["get", "list", "watch", "create", "update", "patch", "delete"],
},
{
apiGroups: [API_GROUP],
resources: ["minecraftservers/status", "velocityproxies/status"],
verbs: ["get", "update", "patch"],
},
],
};
export const minikuraClusterRoleBinding = {
apiVersion: "rbac.authorization.k8s.io/v1",
kind: "ClusterRoleBinding",
metadata: {
name: "minikura-operator-role-binding",
},
subjects: [
{
kind: "ServiceAccount",
name: "minikura-operator",
namespace: NAMESPACE,
},
],
roleRef: {
kind: "ClusterRole",
name: "minikura-operator-role",
apiGroup: "rbac.authorization.k8s.io",
},
};
export const minikuraOperatorDeployment = {
apiVersion: "apps/v1",
kind: "Deployment",
metadata: {
name: "minikura-operator",
namespace: NAMESPACE,
},
spec: {
replicas: 1,
selector: {
matchLabels: {
app: "minikura-operator",
},
},
template: {
metadata: {
labels: {
app: "minikura-operator",
},
},
spec: {
serviceAccountName: "minikura-operator",
containers: [
{
name: "operator",
image: "${REGISTRY_URL}/minikura-operator:latest",
env: [
{
name: "DATABASE_URL",
valueFrom: {
secretKeyRef: {
name: "minikura-operator-secrets",
key: "DATABASE_URL",
},
},
},
{
name: "KUBERNETES_NAMESPACE",
value: NAMESPACE,
},
{
name: "USE_CRDS",
value: "true",
},
],
resources: {
requests: {
memory: "256Mi",
cpu: "200m",
},
limits: {
memory: "512Mi",
cpu: "500m",
},
},
livenessProbe: {
exec: {
command: ["bun", "-e", "console.log('Health check')"],
},
initialDelaySeconds: 30,
periodSeconds: 30,
},
readinessProbe: {
exec: {
command: ["bun", "-e", "console.log('Ready check')"],
},
initialDelaySeconds: 5,
periodSeconds: 10,
},
},
],
},
},
},
};
packages/k8s-operator/src/crds/reverseProxy.ts
-156
View File
@@ -1,156 +0,0 @@
import { API_GROUP, API_VERSION, RESOURCE_TYPES } from "../config/constants";
export const REVERSE_PROXY_SERVER_CRD = {
apiVersion: "apiextensions.k8s.io/v1",
kind: "CustomResourceDefinition",
metadata: {
name: `${RESOURCE_TYPES.REVERSE_PROXY_SERVER.plural}.${API_GROUP}`,
},
spec: {
group: API_GROUP,
versions: [
{
name: API_VERSION,
served: true,
storage: true,
schema: {
openAPIV3Schema: {
type: "object",
properties: {
spec: {
type: "object",
required: ["id", "external_address", "external_port"],
properties: {
id: {
type: "string",
pattern: "^[a-zA-Z0-9-_]+$",
description: "ID of the reverse proxy server",
},
description: {
type: "string",
nullable: true,
description: "Optional description of the server",
},
external_address: {
type: "string",
description: "External address of the proxy server",
},
external_port: {
type: "integer",
minimum: 1,
maximum: 65535,
description: "External port of the proxy server",
},
listen_port: {
type: "integer",
minimum: 1,
maximum: 65535,
default: 25565,
nullable: true,
description: "Port the proxy server listens on internally",
},
type: {
type: "string",
enum: ["VELOCITY", "BUNGEECORD"],
default: "VELOCITY",
nullable: true,
description: "Type of the reverse proxy server",
},
memory: {
type: "string",
default: "512M",
nullable: true,
description: "Memory allocation for the server",
},
environmentVariables: {
type: "array",
nullable: true,
items: {
type: "object",
required: ["key", "value"],
properties: {
key: {
type: "string",
description: "Environment variable key",
},
value: {
type: "string",
description: "Environment variable value",
},
},
},
},
},
},
status: {
type: "object",
nullable: true,
properties: {
phase: {
type: "string",
enum: ["Pending", "Running", "Failed"],
description: "Current phase of the server",
},
message: {
type: "string",
nullable: true,
description: "Detailed message about the current status",
},
apiKey: {
type: "string",
nullable: true,
description: "API key for server communication",
},
internalId: {
type: "string",
nullable: true,
description: "Internal ID assigned by Minikura",
},
lastSyncedAt: {
type: "string",
nullable: true,
description: "Last time the server was synced with Kubernetes",
},
},
},
},
},
},
additionalPrinterColumns: [
{
name: "Type",
type: "string",
jsonPath: ".spec.type",
},
{
name: "External Address",
type: "string",
jsonPath: ".spec.external_address",
},
{
name: "External Port",
type: "integer",
jsonPath: ".spec.external_port",
},
{
name: "Status",
type: "string",
jsonPath: ".status.phase",
},
{
name: "Age",
type: "date",
jsonPath: ".metadata.creationTimestamp",
},
],
},
],
scope: "Namespaced",
names: {
singular: RESOURCE_TYPES.REVERSE_PROXY_SERVER.singular,
plural: RESOURCE_TYPES.REVERSE_PROXY_SERVER.plural,
kind: RESOURCE_TYPES.REVERSE_PROXY_SERVER.kind,
shortNames: RESOURCE_TYPES.REVERSE_PROXY_SERVER.shortNames,
},
},
};
packages/k8s-operator/src/crds/server.ts
-132
View File
@@ -1,132 +0,0 @@
import { API_GROUP, API_VERSION, RESOURCE_TYPES } from "../config/constants";
export const MINECRAFT_SERVER_CRD = {
apiVersion: "apiextensions.k8s.io/v1",
kind: "CustomResourceDefinition",
metadata: {
name: `${RESOURCE_TYPES.MINECRAFT_SERVER.plural}.${API_GROUP}`,
},
spec: {
group: API_GROUP,
versions: [
{
name: API_VERSION,
served: true,
storage: true,
schema: {
openAPIV3Schema: {
type: "object",
properties: {
spec: {
type: "object",
required: ["id", "type", "listen_port"],
properties: {
id: {
type: "string",
pattern: "^[a-zA-Z0-9-_]+$",
description: "ID of the Minecraft server",
},
description: {
type: "string",
nullable: true,
description: "Optional description of the server",
},
listen_port: {
type: "integer",
minimum: 1,
maximum: 65535,
description: "Port the server listens on",
},
type: {
type: "string",
enum: ["STATEFUL", "STATELESS"],
description: "Type of the server",
},
memory: {
type: "string",
nullable: true,
default: "1G",
description: "Memory allocation for the server",
},
environmentVariables: {
type: "array",
nullable: true,
items: {
type: "object",
required: ["key", "value"],
properties: {
key: {
type: "string",
description: "Environment variable key",
},
value: {
type: "string",
description: "Environment variable value",
},
},
},
},
},
},
status: {
type: "object",
nullable: true,
properties: {
phase: {
type: "string",
enum: ["Pending", "Running", "Failed"],
description: "Current phase of the server",
},
message: {
type: "string",
nullable: true,
description: "Detailed message about the current status",
},
apiKey: {
type: "string",
nullable: true,
description: "API key for server communication",
},
internalId: {
type: "string",
nullable: true,
description: "Internal ID assigned by Minikura",
},
lastSyncedAt: {
type: "string",
nullable: true,
description: "Last time the server was synced with Kubernetes",
},
},
},
},
},
},
additionalPrinterColumns: [
{
name: "Type",
type: "string",
jsonPath: ".spec.type",
},
{
name: "Status",
type: "string",
jsonPath: ".status.phase",
},
{
name: "Age",
type: "date",
jsonPath: ".metadata.creationTimestamp",
},
],
},
],
scope: "Namespaced",
names: {
singular: RESOURCE_TYPES.MINECRAFT_SERVER.singular,
plural: RESOURCE_TYPES.MINECRAFT_SERVER.plural,
kind: RESOURCE_TYPES.MINECRAFT_SERVER.kind,
shortNames: RESOURCE_TYPES.MINECRAFT_SERVER.shortNames,
},
},
};
packages/k8s-operator/src/index.ts
-81
View File
@@ -1,81 +0,0 @@
import { dotenvLoad } from "dotenv-mono";
const _dotenv = dotenvLoad();
import { prisma } from "@minikura/db";
import { ENABLE_CRD_REFLECTION, NAMESPACE } from "./config/constants";
import { ReverseProxyController } from "./controllers/reverse-proxy-controller";
import { ServerController } from "./controllers/server-controller";
import { setupCRDRegistration } from "./utils/crd-registrar";
import { KubernetesClient } from "./utils/k8s-client";
import { logger } from "./utils/logger";
async function main() {
logger.info(
{ namespace: NAMESPACE, crdReflection: ENABLE_CRD_REFLECTION },
"Starting Minikura Kubernetes Operator"
);
try {
const k8sClient = KubernetesClient.getInstance();
logger.info({ namespace: NAMESPACE }, "Successfully connected to Kubernetes cluster");
const serverController = new ServerController(prisma, NAMESPACE);
const reverseProxyController = new ReverseProxyController(prisma, NAMESPACE);
serverController.startWatching();
reverseProxyController.startWatching();
if (ENABLE_CRD_REFLECTION) {
logger.info("CRD reflection enabled - will create Custom Resources to mirror database state");
try {
await setupCRDRegistration(prisma, k8sClient, NAMESPACE);
logger.info("CRD registration completed successfully");
} catch (error: any) {
logger.error(
{
err: error,
message: error.message,
statusCode: error.response?.statusCode,
body: error.response?.body,
},
"Failed to setup CRD registration, continuing without CRD reflection"
);
logger.warn(
"Kubernetes resources (Deployments, Services) will still be created, but Custom Resources will not be reflected"
);
}
}
logger.info("Minikura Kubernetes Operator is now running and watching for changes");
process.on("SIGINT", gracefulShutdown);
process.on("SIGTERM", gracefulShutdown);
function gracefulShutdown() {
logger.info("Received shutdown signal, shutting down gracefully");
serverController.stopWatching();
reverseProxyController.stopWatching();
prisma.$disconnect();
logger.info("All resources released, exiting process");
process.exit(0);
}
} catch (error: any) {
logger.fatal(
{
err: error,
message: error.message,
statusCode: error.response?.statusCode,
body: error.response?.body,
stack: error.stack,
},
"Failed to start Minikura Kubernetes Operator"
);
process.exit(1);
}
}
main().catch((error) => {
logger.fatal({ err: error }, "Unhandled error in main process");
process.exit(1);
});
packages/k8s-operator/src/resources/reverseProxyServer.ts
-230
View File
@@ -1,230 +0,0 @@
import type * as k8s from "@kubernetes/client-node";
import type { ReverseProxyServerType } from "@minikura/db";
import { LABEL_PREFIX } from "../config/constants";
import { DEFAULT_PROXY_MEMORY, JAVA_MEMORY_FACTOR } from "../config/resource-defaults";
import type { ReverseProxyConfig } from "../types";
import { logger } from "../utils/logger";
import { calculateJavaMemory, convertToK8sFormat } from "../utils/memory";
import { mapServiceType } from "../utils/service-type";
export async function createReverseProxyServer(
server: ReverseProxyConfig,
appsApi: k8s.AppsV1Api,
coreApi: k8s.CoreV1Api,
_networkingApi: k8s.NetworkingV1Api,
namespace: string
): Promise<void> {
logger.debug(
{ proxyId: server.id, proxyType: server.type, namespace },
"Creating reverse proxy server"
);
const serverType = server.type.toLowerCase();
const serverName = `${serverType}-${server.id}`;
const configMap = {
apiVersion: "v1",
kind: "ConfigMap",
metadata: {
name: `${serverName}-config`,
namespace: namespace,
labels: {
app: serverName,
[`${LABEL_PREFIX}/server-type`]: serverType,
[`${LABEL_PREFIX}/proxy-id`]: server.id,
},
},
data: {
"minikura-api-key": server.apiKey,
},
};
try {
await coreApi.createNamespacedConfigMap({ namespace, body: configMap });
logger.debug({ proxyId: server.id, resource: "ConfigMap" }, "Created ConfigMap");
} catch (error: any) {
if (error.code === 409) {
await coreApi.replaceNamespacedConfigMap({
name: `${serverName}-config`,
namespace,
body: configMap,
});
logger.debug({ proxyId: server.id, resource: "ConfigMap" }, "Updated ConfigMap");
} else {
throw error;
}
}
const service = {
apiVersion: "v1",
kind: "Service",
metadata: {
name: serverName,
namespace: namespace,
labels: {
app: serverName,
[`${LABEL_PREFIX}/server-type`]: serverType,
[`${LABEL_PREFIX}/proxy-id`]: server.id,
},
},
spec: {
selector: {
app: serverName,
},
ports: [
{
port: server.external_port,
targetPort: server.listen_port,
protocol: "TCP",
name: "minecraft",
},
],
type: mapServiceType(server.service_type, "LoadBalancer"),
},
};
try {
await coreApi.createNamespacedService({ namespace, body: service });
logger.debug({ proxyId: server.id, resource: "Service" }, "Created Service");
} catch (error: any) {
if (error.code === 409) {
await coreApi.replaceNamespacedService({ name: serverName, namespace, body: service });
logger.debug({ proxyId: server.id, resource: "Service" }, "Updated Service");
} else {
throw error;
}
}
const deployment = {
apiVersion: "apps/v1",
kind: "Deployment",
metadata: {
name: serverName,
namespace: namespace,
labels: {
app: serverName,
[`${LABEL_PREFIX}/server-type`]: serverType,
[`${LABEL_PREFIX}/proxy-id`]: server.id,
},
},
spec: {
replicas: 1,
selector: {
matchLabels: {
app: serverName,
},
},
template: {
metadata: {
labels: {
app: serverName,
[`${LABEL_PREFIX}/server-type`]: serverType,
[`${LABEL_PREFIX}/proxy-id`]: server.id,
},
},
spec: {
containers: [
{
name: serverType,
image: "itzg/mc-proxy:latest",
ports: [
{
containerPort: server.listen_port,
name: "minecraft",
},
],
env: [
{
name: "TYPE",
value: server.type,
},
{
name: "NETWORKADDRESS_CACHE_TTL",
value: "30",
},
{
name: "MEMORY",
value: calculateJavaMemory(
server.memory || DEFAULT_PROXY_MEMORY,
JAVA_MEMORY_FACTOR
),
},
...(server.env_variables || []).map((ev) => ({
name: ev.key,
value: ev.value,
})),
],
readinessProbe: {
tcpSocket: {
port: server.listen_port,
},
initialDelaySeconds: 30,
periodSeconds: 10,
},
resources: {
requests: {
memory: convertToK8sFormat(server.memory || DEFAULT_PROXY_MEMORY),
cpu: "250m",
},
limits: {
memory: convertToK8sFormat(server.memory || DEFAULT_PROXY_MEMORY),
cpu: "500m",
},
},
},
],
},
},
},
};
try {
await appsApi.createNamespacedDeployment({ namespace, body: deployment });
logger.debug({ proxyId: server.id, resource: "Deployment" }, "Created Deployment");
} catch (error: any) {
if (error.code === 409) {
await appsApi.replaceNamespacedDeployment({ name: serverName, namespace, body: deployment });
logger.debug({ proxyId: server.id, resource: "Deployment" }, "Updated Deployment");
} else {
throw error;
}
}
}
export async function deleteReverseProxyServer(
proxyId: string,
proxyType: ReverseProxyServerType,
appsApi: k8s.AppsV1Api,
coreApi: k8s.CoreV1Api,
namespace: string
): Promise<void> {
const serverType = proxyType.toLowerCase();
const name = `${serverType}-${proxyId}`;
try {
await appsApi.deleteNamespacedDeployment({ name, namespace });
logger.debug({ proxyId, resource: "Deployment" }, "Deleted Deployment");
} catch (error: any) {
if (error.response?.statusCode !== 404) {
logger.error({ err: error, proxyId, resource: "Deployment" }, "Failed to delete Deployment");
}
}
try {
await coreApi.deleteNamespacedService({ name, namespace });
logger.debug({ proxyId, resource: "Service" }, "Deleted Service");
} catch (error: any) {
if (error.response?.statusCode !== 404) {
logger.error({ err: error, proxyId, resource: "Service" }, "Failed to delete Service");
}
}
try {
await coreApi.deleteNamespacedConfigMap({ name: `${name}-config`, namespace });
logger.debug({ proxyId, resource: "ConfigMap" }, "Deleted ConfigMap");
} catch (error: any) {
if (error.response?.statusCode !== 404) {
logger.error({ err: error, proxyId, resource: "ConfigMap" }, "Failed to delete ConfigMap");
}
}
}
packages/k8s-operator/src/resources/server.ts
-426
View File
@@ -1,426 +0,0 @@
import type * as k8s from "@kubernetes/client-node";
import { ServerType } from "@minikura/db";
import { LABEL_PREFIX } from "../config/constants";
import { DEFAULT_SERVER_MEMORY, JAVA_MEMORY_FACTOR } from "../config/resource-defaults";
import type { ServerConfig } from "../types";
import { logger } from "../utils/logger";
import { calculateJavaMemory, convertToK8sFormat } from "../utils/memory";
import { mapServiceType } from "../utils/service-type";
export async function createServer(
server: ServerConfig,
appsApi: k8s.AppsV1Api,
coreApi: k8s.CoreV1Api,
_networkingApi: k8s.NetworkingV1Api,
namespace: string
): Promise<void> {
const serverName = `minecraft-${server.id}`;
const configMap = {
apiVersion: "v1",
kind: "ConfigMap",
metadata: {
name: `${serverName}-config`,
namespace: namespace,
labels: {
app: serverName,
[`${LABEL_PREFIX}/server-type`]: server.type.toLowerCase(),
[`${LABEL_PREFIX}/server-id`]: server.id,
},
},
data: {
"server-type": server.type,
"minikura-api-key": server.apiKey,
},
};
try {
await coreApi.createNamespacedConfigMap({ namespace, body: configMap });
logger.debug({ serverId: server.id, resource: "ConfigMap" }, "Created ConfigMap");
} catch (err: any) {
if (err.code === 409) {
await coreApi.replaceNamespacedConfigMap({
name: `${serverName}-config`,
namespace,
body: configMap,
});
logger.debug({ serverId: server.id, resource: "ConfigMap" }, "Updated ConfigMap");
} else {
throw err;
}
}
const service = {
apiVersion: "v1",
kind: "Service",
metadata: {
name: serverName,
namespace: namespace,
labels: {
app: serverName,
[`${LABEL_PREFIX}/server-type`]: server.type.toLowerCase(),
[`${LABEL_PREFIX}/server-id`]: server.id,
},
},
spec: {
selector: {
app: serverName,
},
ports: [
{
port: server.listen_port,
targetPort: 25565,
protocol: "TCP",
name: "minecraft",
},
],
type: mapServiceType(server.service_type),
},
};
try {
await coreApi.createNamespacedService({ namespace, body: service });
logger.debug(
{ serverId: server.id, resource: "Service", port: server.listen_port },
"Created Service"
);
} catch (err: any) {
if (err.code === 409) {
await coreApi.replaceNamespacedService({ name: serverName, namespace, body: service });
logger.debug({ serverId: server.id, resource: "Service" }, "Updated Service");
} else {
throw err;
}
}
if (server.type === ServerType.STATELESS) {
await createDeployment(serverName, server, appsApi, namespace);
} else {
await createStatefulSet(serverName, server, appsApi, namespace);
}
}
async function createDeployment(
serverName: string,
server: ServerConfig,
appsApi: k8s.AppsV1Api,
namespace: string
): Promise<void> {
const deployment = {
apiVersion: "apps/v1",
kind: "Deployment",
metadata: {
name: serverName,
namespace: namespace,
labels: {
app: serverName,
[`${LABEL_PREFIX}/server-type`]: "stateless",
[`${LABEL_PREFIX}/server-id`]: server.id,
},
},
spec: {
replicas: 1,
selector: {
matchLabels: {
app: serverName,
},
},
template: {
metadata: {
labels: {
app: serverName,
[`${LABEL_PREFIX}/server-type`]: "stateless",
[`${LABEL_PREFIX}/server-id`]: server.id,
},
},
spec: {
containers: [
{
name: "minecraft",
image: "itzg/minecraft-server",
ports: [
{
containerPort: 25565,
name: "minecraft",
},
],
env: [
{
name: "EULA",
value: "TRUE",
},
{
name: "TYPE",
value: "VANILLA",
},
{
name: "MEMORY",
value: calculateJavaMemory(
server.memory || DEFAULT_SERVER_MEMORY,
JAVA_MEMORY_FACTOR
),
},
{
name: "OPS",
value: "",
},
{
name: "OVERRIDE_SERVER_PROPERTIES",
value: "true",
},
{
name: "ENABLE_RCON",
value: "false",
},
...(server.env_variables || []).map((ev) => ({
name: ev.key,
value: ev.value,
})),
],
volumeMounts: [
{
name: "config",
mountPath: "/config",
},
],
readinessProbe: {
tcpSocket: {
port: 25565,
},
initialDelaySeconds: 30,
periodSeconds: 10,
},
resources: {
requests: {
memory: convertToK8sFormat(server.memory || DEFAULT_SERVER_MEMORY),
cpu: "250m",
},
limits: {
memory: convertToK8sFormat(server.memory || DEFAULT_SERVER_MEMORY),
cpu: "500m",
},
},
},
],
volumes: [
{
name: "config",
configMap: {
name: `${serverName}-config`,
},
},
],
},
},
},
};
try {
await appsApi.createNamespacedDeployment({ namespace, body: deployment });
logger.debug({ serverId: server.id, resource: "Deployment" }, "Created Deployment");
} catch (err: any) {
if (err.code === 409) {
await appsApi.replaceNamespacedDeployment({ name: serverName, namespace, body: deployment });
logger.debug({ serverId: server.id, resource: "Deployment" }, "Updated Deployment");
} else {
throw err;
}
}
}
async function createStatefulSet(
serverName: string,
server: ServerConfig,
appsApi: k8s.AppsV1Api,
namespace: string
): Promise<void> {
const statefulSet = {
apiVersion: "apps/v1",
kind: "StatefulSet",
metadata: {
name: serverName,
namespace: namespace,
labels: {
app: serverName,
[`${LABEL_PREFIX}/server-type`]: "stateful",
[`${LABEL_PREFIX}/server-id`]: server.id,
},
},
spec: {
serviceName: serverName,
replicas: 1,
selector: {
matchLabels: {
app: serverName,
},
},
template: {
metadata: {
labels: {
app: serverName,
[`${LABEL_PREFIX}/server-type`]: "stateful",
[`${LABEL_PREFIX}/server-id`]: server.id,
},
},
spec: {
containers: [
{
name: "minecraft",
image: "itzg/minecraft-server",
ports: [
{
containerPort: 25565,
name: "minecraft",
},
],
env: [
{
name: "EULA",
value: "TRUE",
},
{
name: "TYPE",
value: "VANILLA",
},
{
name: "MEMORY",
value: calculateJavaMemory(
server.memory || DEFAULT_SERVER_MEMORY,
JAVA_MEMORY_FACTOR
),
},
{
name: "OPS",
value: "",
},
{
name: "OVERRIDE_SERVER_PROPERTIES",
value: "true",
},
{
name: "ENABLE_RCON",
value: "false",
},
...(server.env_variables || []).map((ev) => ({
name: ev.key,
value: ev.value,
})),
],
volumeMounts: [
{
name: "data",
mountPath: "/data",
},
{
name: "config",
mountPath: "/config",
},
],
readinessProbe: {
tcpSocket: {
port: 25565,
},
initialDelaySeconds: 60,
periodSeconds: 10,
},
resources: {
requests: {
memory: convertToK8sFormat(server.memory),
cpu: "250m",
},
limits: {
memory: convertToK8sFormat(server.memory),
cpu: "500m",
},
},
},
],
volumes: [
{
name: "config",
configMap: {
name: `${serverName}-config`,
},
},
],
},
},
volumeClaimTemplates: [
{
metadata: {
name: "data",
},
spec: {
accessModes: ["ReadWriteOnce"],
resources: {
requests: {
storage: "1Gi",
},
},
},
},
],
},
};
try {
await appsApi.createNamespacedStatefulSet({ namespace, body: statefulSet });
logger.debug({ serverId: server.id, resource: "StatefulSet" }, "Created StatefulSet");
} catch (err: any) {
if (err.code === 409) {
await appsApi.replaceNamespacedStatefulSet({
name: serverName,
namespace,
body: statefulSet,
});
logger.debug({ serverId: server.id, resource: "StatefulSet" }, "Updated StatefulSet");
} else {
throw err;
}
}
}
export async function deleteServer(
serverId: string,
appsApi: k8s.AppsV1Api,
coreApi: k8s.CoreV1Api,
namespace: string
): Promise<void> {
const serverName = `minecraft-${serverId}`;
try {
await appsApi.deleteNamespacedDeployment({ name: serverName, namespace });
logger.debug({ serverName, resource: "Deployment" }, "Deleted Deployment");
} catch (err: any) {
if (err.response?.statusCode !== 404) {
logger.error({ err, serverName, resource: "Deployment" }, "Failed to delete Deployment");
}
}
try {
await appsApi.deleteNamespacedStatefulSet({ name: serverName, namespace });
logger.debug({ serverName, resource: "StatefulSet" }, "Deleted StatefulSet");
} catch (err: any) {
if (err.response?.statusCode !== 404) {
logger.error({ err, serverName, resource: "StatefulSet" }, "Failed to delete StatefulSet");
}
}
try {
await coreApi.deleteNamespacedService({ name: serverName, namespace });
logger.debug({ serverName, resource: "Service" }, "Deleted Service");
} catch (err: any) {
if (err.response?.statusCode !== 404) {
logger.error({ err, serverName, resource: "Service" }, "Failed to delete Service");
}
}
try {
await coreApi.deleteNamespacedConfigMap({ name: `${serverName}-config`, namespace });
logger.debug({ serverName, resource: "ConfigMap" }, "Deleted ConfigMap");
} catch (err: any) {
if (err.response?.statusCode !== 404) {
logger.error({ err, serverName, resource: "ConfigMap" }, "Failed to delete ConfigMap");
}
}
}
packages/k8s-operator/src/scripts/apply-crds.ts
-37
View File
@@ -1,37 +0,0 @@
import { PrismaClient } from "@minikura/db";
import { dotenvLoad } from "dotenv-mono";
import { NAMESPACE } from "../config/constants";
import { setupCRDRegistration } from "../utils/crd-registrar";
import { KubernetesClient } from "../utils/k8s-client";
import { registerRBACResources } from "../utils/rbac-registrar";
dotenvLoad();
async function main() {
console.log("Starting to apply TypeScript-defined CRDs to Kubernetes cluster...");
try {
const k8sClient = KubernetesClient.getInstance();
console.log(`Connected to Kubernetes cluster, using namespace: ${NAMESPACE}`);
await registerRBACResources(k8sClient);
console.log("Registering Custom Resource Definitions...");
const prisma = new PrismaClient();
await setupCRDRegistration(prisma, k8sClient, NAMESPACE);
console.log("Successfully applied all resources to Kubernetes cluster");
process.exit(0);
} catch (error: any) {
console.error("Failed to apply resources:", error.message);
if (error.stack) {
console.error(error.stack);
}
process.exit(1);
}
}
main().catch((error) => {
console.error("Unhandled error:", error);
process.exit(1);
});
packages/k8s-operator/src/services/notification.service.ts
-80
View File
@@ -1,80 +0,0 @@
import { createLogger } from "@minikura/shared";
import pg from "pg";
const logger = createLogger("notification-service");
export class NotificationService {
private pgClient: pg.Client | null = null;
private handlers = new Map<string, Set<(payload: unknown) => void | Promise<void>>>();
async connect(connectionString: string): Promise<void> {
if (!connectionString) {
throw new Error("Database connection string is required");
}
logger.info("Connecting to PostgreSQL");
this.pgClient = new pg.Client({ connectionString });
await this.pgClient.connect();
this.pgClient.on("notification", async (msg) => {
const handlers = this.handlers.get(msg.channel);
if (!handlers) return;
try {
const payload = msg.payload ? JSON.parse(msg.payload) : {};
logger.info({ channel: msg.channel, payload }, "Received notification");
for (const handler of handlers) {
try {
await handler(payload);
} catch (err) {
logger.error({ err, channel: msg.channel }, "Error in notification handler");
}
}
} catch (err) {
logger.error({ err }, "Failed to parse notification payload");
}
});
logger.info("Connected to PostgreSQL successfully");
}
async listen(
channel: string,
handler: (payload: unknown) => void | Promise<void>
): Promise<void> {
if (!this.pgClient) {
throw new Error("NotificationService not connected");
}
if (!this.handlers.has(channel)) {
this.handlers.set(channel, new Set());
await this.pgClient.query(`LISTEN ${channel}`);
logger.info({ channel }, "Listening on channel");
}
this.handlers.get(channel)?.add(handler);
}
async unlisten(channel: string): Promise<void> {
if (!this.pgClient) return;
this.handlers.delete(channel);
await this.pgClient.query(`UNLISTEN ${channel}`);
logger.info({ channel }, "Stopped listening on channel");
}
async disconnect(): Promise<void> {
if (!this.pgClient) return;
logger.info("Disconnecting from PostgreSQL");
await this.pgClient.end();
this.pgClient = null;
this.handlers.clear();
logger.info("Disconnected from PostgreSQL");
}
isConnected(): boolean {
return this.pgClient !== null;
}
}
packages/k8s-operator/src/types/index.ts
-87
View File
@@ -1,87 +0,0 @@
import type {
CustomEnvironmentVariable,
ReverseProxyServer as PrismaReverseProxyServer,
Server as PrismaServer,
} from "@minikura/db";
export interface CustomResource {
apiVersion: string;
kind: string;
metadata: {
name: string;
namespace?: string;
labels?: Record<string, string>;
annotations?: Record<string, string>;
[key: string]: any;
};
}
export type ServerConfig = Pick<
PrismaServer,
"id" | "description" | "type" | "listen_port" | "memory" | "service_type"
> & {
apiKey: string;
env_variables?: Array<Pick<CustomEnvironmentVariable, "key" | "value">>;
};
export type MinecraftServerSpec = Pick<
PrismaServer,
"id" | "description" | "type" | "listen_port" | "memory"
> & {
environmentVariables?: Array<Pick<CustomEnvironmentVariable, "key" | "value">>;
};
export interface MinecraftServerStatus {
phase: "Pending" | "Running" | "Failed";
message?: string;
apiKey?: string;
internalId?: string;
lastSyncedAt?: string;
}
export interface MinecraftServerCRD extends CustomResource {
spec: MinecraftServerSpec;
status?: MinecraftServerStatus;
}
export type ReverseProxyConfig = Pick<
PrismaReverseProxyServer,
| "id"
| "description"
| "external_address"
| "external_port"
| "listen_port"
| "type"
| "memory"
| "service_type"
> & {
apiKey: string;
env_variables?: Array<Pick<CustomEnvironmentVariable, "key" | "value">>;
};
export type ReverseProxyServerSpec = Partial<
Pick<
PrismaReverseProxyServer,
"id" | "description" | "external_address" | "external_port" | "listen_port" | "type" | "memory"
>
> & {
id: string;
external_address: string;
external_port: number;
environmentVariables?: Array<Pick<CustomEnvironmentVariable, "key" | "value">>;
};
export interface ReverseProxyServerStatus {
phase: "Pending" | "Running" | "Failed";
message?: string;
apiKey?: string;
internalId?: string;
lastSyncedAt?: string;
}
export interface ReverseProxyServerCRD extends CustomResource {
spec: ReverseProxyServerSpec;
status?: ReverseProxyServerStatus;
}
export type EnvironmentVariable = Pick<CustomEnvironmentVariable, "key" | "value">;
packages/k8s-operator/src/types/k8s-types.ts
-25
View File
@@ -1,25 +0,0 @@
import type * as k8s from "@kubernetes/client-node";
export interface K8sApiError extends Error {
code?: number;
body?: string;
headers?: Record<string, string>;
}
export interface CustomResourceResponse<T = unknown> {
metadata?: k8s.V1ObjectMeta;
spec?: T;
status?: Record<string, unknown>;
body?: CustomResourceResponse<T>;
}
export interface CustomResourceListResponse<T = unknown> {
items?: CustomResourceResponse<T>[];
body?: {
items?: CustomResourceResponse<T>[];
};
}
export function isK8sApiError(error: unknown): error is K8sApiError {
return error instanceof Error && ("code" in error || "body" in error || "headers" in error);
}
packages/k8s-operator/src/utils/crd-registrar.ts
-633
View File
@@ -1,633 +0,0 @@
import type * as k8s from "@kubernetes/client-node";
import type { PrismaClient } from "@minikura/db";
import { API_GROUP, API_VERSION, LABEL_PREFIX } from "../config/constants";
import { REVERSE_PROXY_SERVER_CRD } from "../crds/reverseProxy";
import { MINECRAFT_SERVER_CRD } from "../crds/server";
import type { CustomResourceListResponse, CustomResourceResponse } from "../types/k8s-types";
import { isK8sApiError } from "../types/k8s-types";
import type { KubernetesClient } from "./k8s-client";
import { logger } from "./logger";
async function retryWithBackoff<T>(
operation: () => Promise<T>,
options: {
maxRetries?: number;
initialDelay?: number;
maxDelay?: number;
operationName?: string;
} = {}
): Promise<T> {
const {
maxRetries = 5,
initialDelay = 1000,
maxDelay = 10000,
operationName = "operation",
} = options;
let lastError: Error | undefined;
for (let attempt = 0; attempt <= maxRetries; attempt++) {
try {
return await operation();
} catch (error: unknown) {
lastError = error as Error;
const is429 = isK8sApiError(error) && error.code === 429;
const isStorageInitializing =
isK8sApiError(error) && error.body?.includes("storage is (re)initializing");
if (!is429 && !isStorageInitializing) {
throw error;
}
if (attempt === maxRetries) {
logger.error({ operationName, maxRetries }, "Operation failed after max retries");
throw error;
}
let delay = initialDelay * 2 ** attempt;
if (isK8sApiError(error) && error.headers?.["retry-after"]) {
const retryAfter = parseInt(error.headers["retry-after"], 10);
if (!Number.isNaN(retryAfter)) {
delay = retryAfter * 1000;
}
}
delay = Math.min(delay, maxDelay);
const errorMessage = error instanceof Error ? error.message : String(error);
logger.warn(
{
operationName,
attempt: attempt + 1,
maxAttempts: maxRetries + 1,
delayMs: delay,
errorMessage,
},
"Operation failed, retrying with backoff"
);
await new Promise((resolve) => setTimeout(resolve, delay));
}
}
throw lastError;
}
export async function setupCRDRegistration(
prisma: PrismaClient,
k8sClient: KubernetesClient,
namespace: string
): Promise<void> {
await registerCRDs(k8sClient);
logger.info("Waiting for Kubernetes storage to stabilize after CRD registration");
await new Promise((resolve) => setTimeout(resolve, 2000));
await startCRDReflector(prisma, k8sClient, namespace);
}
async function registerCRDs(k8sClient: KubernetesClient): Promise<void> {
try {
const apiExtensionsClient = k8sClient.getApiExtensionsApi();
logger.info(
{ apiGroup: API_GROUP, apiVersion: API_VERSION },
"Registering Custom Resource Definitions"
);
try {
await apiExtensionsClient.createCustomResourceDefinition({ body: MINECRAFT_SERVER_CRD });
logger.info(
{ crd: "MinecraftServer", apiGroup: API_GROUP, apiVersion: API_VERSION },
"CRD created successfully"
);
} catch (error: any) {
if (error.code === 409) {
logger.debug("MinecraftServer CRD already exists, skipping creation");
} else {
logger.error({ err: error }, "Failed to create MinecraftServer CRD");
throw error;
}
}
try {
await apiExtensionsClient.createCustomResourceDefinition({ body: REVERSE_PROXY_SERVER_CRD });
logger.info(
{ crd: "ReverseProxyServer", apiGroup: API_GROUP, apiVersion: API_VERSION },
"CRD created successfully"
);
} catch (error: any) {
if (error.code === 409) {
logger.debug("ReverseProxyServer CRD already exists, skipping creation");
} else {
logger.error({ err: error }, "Failed to create ReverseProxyServer CRD");
throw error;
}
}
} catch (error) {
logger.error({ err: error }, "Failed to register CRDs");
throw error;
}
}
async function startCRDReflector(
prisma: PrismaClient,
k8sClient: KubernetesClient,
namespace: string
): Promise<void> {
const customObjectsApi = k8sClient.getCustomObjectsApi();
const reflectedMinecraftServers = new Map<string, string>();
const reflectedReverseProxyServers = new Map<string, string>();
logger.info("Starting CRD reflector to sync database state to custom resources");
await syncDBtoCRDs(
prisma,
customObjectsApi,
namespace,
reflectedMinecraftServers,
reflectedReverseProxyServers
);
setInterval(async () => {
await syncDBtoCRDs(
prisma,
customObjectsApi,
namespace,
reflectedMinecraftServers,
reflectedReverseProxyServers
);
}, 30 * 1000);
}
async function syncDBtoCRDs(
prisma: PrismaClient,
customObjectsApi: k8s.CustomObjectsApi,
namespace: string,
reflectedMinecraftServers: Map<string, string>,
reflectedReverseProxyServers: Map<string, string>
): Promise<void> {
try {
logger.debug("Starting CRD sync operation");
await syncMinecraftServers(prisma, customObjectsApi, namespace, reflectedMinecraftServers);
await syncReverseProxyServers(
prisma,
customObjectsApi,
namespace,
reflectedReverseProxyServers
);
logger.debug("CRD sync operation completed successfully");
} catch (error) {
logger.error({ err: error }, "Failed to sync database to CRDs");
}
}
async function syncMinecraftServers(
prisma: PrismaClient,
customObjectsApi: k8s.CustomObjectsApi,
namespace: string,
reflectedMinecraftServers: Map<string, string>
): Promise<void> {
try {
const servers = await prisma.server.findMany();
let existingCRs: any[] = [];
try {
const response = await retryWithBackoff(
() =>
customObjectsApi.listNamespacedCustomObject({
group: API_GROUP,
version: API_VERSION,
namespace,
plural: "minecraftservers",
}),
{
maxRetries: 5,
initialDelay: 1000,
operationName: "List MinecraftServer CRs",
}
);
const listResponse = response as unknown as CustomResourceListResponse;
existingCRs = listResponse.body?.items || listResponse.items || [];
} catch (error) {
logger.error(
{ err: error },
"Failed to list MinecraftServer custom resources, assuming none exist"
);
existingCRs = [];
}
const existingCRMap = new Map<string, string>();
const crResourceVersions = new Map<string, string>();
for (const cr of existingCRs) {
const internalId = cr.status?.internalId;
if (internalId) {
existingCRMap.set(internalId, cr.metadata.name);
if (cr.metadata?.resourceVersion) {
crResourceVersions.set(cr.metadata.name, cr.metadata.resourceVersion);
}
}
}
reflectedMinecraftServers.clear();
for (const server of servers) {
const crName = existingCRMap.get(server.id) || `${server.id.toLowerCase()}`;
const serverCR: {
apiVersion: string;
kind: string;
metadata: {
name: string;
namespace: string;
annotations: Record<string, string>;
resourceVersion?: string;
};
spec: any;
status: any;
} = {
apiVersion: `${API_GROUP}/${API_VERSION}`,
kind: "MinecraftServer",
metadata: {
name: crName,
namespace: namespace,
annotations: {
[`${LABEL_PREFIX}/database-managed`]: "true",
[`${LABEL_PREFIX}/last-synced`]: new Date().toISOString(),
},
},
spec: {
id: server.id,
description: server.description,
listen_port: server.listen_port,
type: server.type,
memory: `${server.memory}M`,
},
status: {
phase: "Running",
message: "Managed by database",
internalId: server.id,
apiKey: "[REDACTED]",
lastSyncedAt: new Date().toISOString(),
},
};
try {
const existingCRName = existingCRMap.get(server.id);
if (existingCRName) {
try {
const existingResource = await customObjectsApi.getNamespacedCustomObject({
group: API_GROUP,
version: API_VERSION,
namespace,
plural: "minecraftservers",
name: existingCRName,
});
const resourceResponse = existingResource as CustomResourceResponse;
const resource = resourceResponse.body || resourceResponse;
if (resource?.metadata?.resourceVersion) {
serverCR.metadata.resourceVersion = resource.metadata.resourceVersion;
}
await customObjectsApi.replaceNamespacedCustomObject({
group: API_GROUP,
version: API_VERSION,
namespace,
plural: "minecraftservers",
name: existingCRName,
body: serverCR,
});
logger.debug(
{ crName: existingCRName, serverId: server.id },
"Updated MinecraftServer custom resource"
);
} catch (error) {
logger.error(
{ err: error, serverId: server.id },
"Failed to get/update MinecraftServer custom resource"
);
}
} else {
try {
await customObjectsApi.createNamespacedCustomObject({
group: API_GROUP,
version: API_VERSION,
namespace,
plural: "minecraftservers",
body: serverCR,
});
logger.debug(
{ crName, serverId: server.id },
"Created MinecraftServer custom resource"
);
} catch (createError: any) {
if (createError.code === 409) {
logger.debug({ crName }, "MinecraftServer CR already exists, updating instead");
try {
const existingResource = await customObjectsApi.getNamespacedCustomObject({
group: API_GROUP,
version: API_VERSION,
namespace,
plural: "minecraftservers",
name: crName,
});
const resourceResponse = existingResource as CustomResourceResponse;
let resource = resourceResponse.body || resourceResponse;
if (!resource?.metadata && resourceResponse.metadata) {
resource = resourceResponse;
}
if (resource?.metadata?.resourceVersion) {
serverCR.metadata.resourceVersion = resource.metadata.resourceVersion;
await customObjectsApi.replaceNamespacedCustomObject({
group: API_GROUP,
version: API_VERSION,
namespace,
plural: "minecraftservers",
name: crName,
body: serverCR,
});
logger.debug(
{ crName, serverId: server.id },
"Updated existing MinecraftServer custom resource"
);
} else {
logger.error({ crName }, "Cannot update CR: no resourceVersion in response");
}
} catch (updateError) {
logger.error(
{ err: updateError, crName },
"Failed to update MinecraftServer custom resource"
);
throw updateError;
}
} else {
throw createError;
}
}
}
reflectedMinecraftServers.set(server.id, crName);
} catch (error) {
logger.error(
{ err: error, serverId: server.id },
"Failed to create/update MinecraftServer custom resource"
);
}
}
for (const [dbId, crName] of existingCRMap.entries()) {
if (!servers.some((s) => s.id === dbId)) {
try {
await customObjectsApi.deleteNamespacedCustomObject({
group: API_GROUP,
version: API_VERSION,
namespace,
plural: "minecraftservers",
name: crName,
});
logger.info(
{ crName, serverId: dbId },
"Deleted MinecraftServer CR for removed database record"
);
} catch (error) {
logger.error({ err: error, crName }, "Failed to delete MinecraftServer custom resource");
}
}
}
} catch (error) {
logger.error({ err: error }, "Failed to sync Minecraft servers to custom resources");
}
}
async function syncReverseProxyServers(
prisma: PrismaClient,
customObjectsApi: any,
namespace: string,
reflectedReverseProxyServers: Map<string, string>
): Promise<void> {
try {
const proxies = await prisma.reverseProxyServer.findMany({
include: {
env_variables: true,
},
});
let existingCRs: any[] = [];
try {
const response = await retryWithBackoff(
() =>
customObjectsApi.listNamespacedCustomObject({
group: API_GROUP,
version: API_VERSION,
namespace,
plural: "reverseproxyservers",
}),
{
maxRetries: 5,
initialDelay: 1000,
operationName: "List ReverseProxyServer CRs",
}
);
const listResponse = response as unknown as CustomResourceListResponse;
existingCRs = listResponse.body?.items || listResponse.items || [];
} catch (error) {
logger.error(
{ err: error },
"Failed to list ReverseProxyServer custom resources, assuming none exist"
);
existingCRs = [];
}
const existingCRMap = new Map<string, string>();
const crResourceVersions = new Map<string, string>();
for (const cr of existingCRs) {
const internalId = cr.status?.internalId;
if (internalId) {
existingCRMap.set(internalId, cr.metadata.name);
if (cr.metadata?.resourceVersion) {
crResourceVersions.set(cr.metadata.name, cr.metadata.resourceVersion);
}
}
}
reflectedReverseProxyServers.clear();
for (const proxy of proxies) {
const crName = existingCRMap.get(proxy.id) || `${proxy.id.toLowerCase()}`;
const proxyCR: {
apiVersion: string;
kind: string;
metadata: {
name: string;
namespace: string;
annotations: Record<string, string>;
resourceVersion?: string;
};
spec: any;
status: any;
} = {
apiVersion: `${API_GROUP}/${API_VERSION}`,
kind: "ReverseProxyServer",
metadata: {
name: crName,
namespace: namespace,
annotations: {
[`${LABEL_PREFIX}/database-managed`]: "true",
[`${LABEL_PREFIX}/last-synced`]: new Date().toISOString(),
},
},
spec: {
id: proxy.id,
description: proxy.description,
external_address: proxy.external_address,
external_port: proxy.external_port,
listen_port: proxy.listen_port,
type: proxy.type,
memory: `${proxy.memory}M`,
environmentVariables: proxy.env_variables?.map((ev) => ({
key: ev.key,
value: ev.value,
})),
},
status: {
phase: "Running",
message: "Managed by database",
internalId: proxy.id,
apiKey: "[REDACTED]",
lastSyncedAt: new Date().toISOString(),
},
};
try {
const existingCRName = existingCRMap.get(proxy.id);
if (existingCRName) {
try {
const existingResource = await customObjectsApi.getNamespacedCustomObject({
group: API_GROUP,
version: API_VERSION,
namespace,
plural: "reverseproxyservers",
name: existingCRName,
});
const resourceResponse = existingResource as CustomResourceResponse;
const resource = resourceResponse.body || resourceResponse;
if (resource?.metadata?.resourceVersion) {
proxyCR.metadata.resourceVersion = resource.metadata.resourceVersion;
}
await customObjectsApi.replaceNamespacedCustomObject({
group: API_GROUP,
version: API_VERSION,
namespace,
plural: "reverseproxyservers",
name: existingCRName,
body: proxyCR,
});
logger.debug(
{ crName: existingCRName, proxyId: proxy.id },
"Updated ReverseProxyServer custom resource"
);
} catch (error) {
logger.error(
{ err: error, proxyId: proxy.id },
"Failed to get/update ReverseProxyServer custom resource"
);
}
} else {
try {
await customObjectsApi.createNamespacedCustomObject({
group: API_GROUP,
version: API_VERSION,
namespace,
plural: "reverseproxyservers",
body: proxyCR,
});
logger.debug(
{ crName, proxyId: proxy.id },
"Created ReverseProxyServer custom resource"
);
} catch (createError: any) {
if (createError.code === 409) {
logger.debug({ crName }, "ReverseProxyServer CR already exists, updating instead");
try {
const existingResource = await customObjectsApi.getNamespacedCustomObject({
group: API_GROUP,
version: API_VERSION,
namespace,
plural: "reverseproxyservers",
name: crName,
});
const resource = existingResource.body as any;
if (resource?.metadata?.resourceVersion) {
proxyCR.metadata.resourceVersion = resource.metadata.resourceVersion;
}
await customObjectsApi.replaceNamespacedCustomObject({
group: API_GROUP,
version: API_VERSION,
namespace,
plural: "reverseproxyservers",
name: crName,
body: proxyCR,
});
logger.debug(
{ crName, proxyId: proxy.id },
"Updated existing ReverseProxyServer custom resource"
);
} catch (updateError) {
logger.error(
{ err: updateError, crName },
"Failed to update ReverseProxyServer custom resource"
);
throw updateError;
}
} else {
throw createError;
}
}
}
reflectedReverseProxyServers.set(proxy.id, crName);
} catch (error) {
logger.error(
{ err: error, proxyId: proxy.id },
"Failed to create/update ReverseProxyServer custom resource"
);
}
}
for (const [dbId, crName] of existingCRMap.entries()) {
if (!proxies.some((p) => p.id === dbId)) {
try {
await customObjectsApi.deleteNamespacedCustomObject({
group: API_GROUP,
version: API_VERSION,
namespace,
plural: "reverseproxyservers",
name: crName,
});
logger.info(
{ crName, proxyId: dbId },
"Deleted ReverseProxyServer CR for removed database record"
);
} catch (error) {
logger.error(
{ err: error, crName },
"Failed to delete ReverseProxyServer custom resource"
);
}
}
}
} catch (error) {
logger.error({ err: error }, "Failed to sync reverse proxy servers to custom resources");
}
}
packages/k8s-operator/src/utils/k8s-client.ts
-71
View File
@@ -1,71 +0,0 @@
import * as k8s from "@kubernetes/client-node";
import { buildKubeConfig } from "@minikura/shared/kube-auth";
import { logger } from "./logger";
export class KubernetesClient {
private static instance: KubernetesClient;
private kc: k8s.KubeConfig;
private appsApi!: k8s.AppsV1Api;
private coreApi!: k8s.CoreV1Api;
private networkingApi!: k8s.NetworkingV1Api;
private customObjectsApi!: k8s.CustomObjectsApi;
private apiExtensionsApi!: k8s.ApiextensionsV1Api;
private constructor() {
this.kc = buildKubeConfig();
this.initializeClients();
}
static getInstance(): KubernetesClient {
if (!KubernetesClient.instance) {
KubernetesClient.instance = new KubernetesClient();
}
return KubernetesClient.instance;
}
private initializeClients(): void {
this.appsApi = this.kc.makeApiClient(k8s.AppsV1Api);
this.coreApi = this.kc.makeApiClient(k8s.CoreV1Api);
this.networkingApi = this.kc.makeApiClient(k8s.NetworkingV1Api);
this.customObjectsApi = this.kc.makeApiClient(k8s.CustomObjectsApi);
this.apiExtensionsApi = this.kc.makeApiClient(k8s.ApiextensionsV1Api);
}
getKubeConfig(): k8s.KubeConfig {
return this.kc;
}
getAppsApi(): k8s.AppsV1Api {
return this.appsApi;
}
getCoreApi(): k8s.CoreV1Api {
return this.coreApi;
}
getNetworkingApi(): k8s.NetworkingV1Api {
return this.networkingApi;
}
getCustomObjectsApi(): k8s.CustomObjectsApi {
return this.customObjectsApi;
}
getApiExtensionsApi(): k8s.ApiextensionsV1Api {
return this.apiExtensionsApi;
}
async handleApiError(error: any, context: string): Promise<never> {
logger.error(
{
context,
message: error?.message,
statusCode: error?.response?.statusCode,
body: error?.response?.body,
},
"Kubernetes API error"
);
throw error;
}
}
packages/k8s-operator/src/utils/logger.ts
-5
View File
@@ -1,5 +0,0 @@
import { createLogger } from "@minikura/shared";
export { createLogger };
export const logger = createLogger("k8s-operator");
packages/k8s-operator/src/utils/memory.ts
-32
View File
@@ -1,32 +0,0 @@
export function calculateJavaMemory(memory: number | string, factor: number): string {
if (typeof memory === "number") {
const calculatedValue = Math.round(memory * factor);
return `${calculatedValue}M`;
}
const match = memory.match(/^(\d+)([MG])$/i);
if (!match) return "512M";
const [, valueStr, unit] = match;
const value = parseInt(valueStr, 10);
const calculatedValue = Math.round(value * factor);
return `${calculatedValue}${unit.toUpperCase()}`;
}
export function convertToK8sFormat(memory: number | string): string {
if (typeof memory === "number") {
return `${memory}Mi`;
}
const match = memory.match(/^(\d+)([MG])$/i);
if (!match) return "1Gi";
const [, valueStr, unit] = match;
if (unit.toUpperCase() === "G") {
return `${valueStr}Gi`;
} else {
return `${valueStr}Mi`;
}
}
packages/k8s-operator/src/utils/rbac-registrar.ts
-203
View File
@@ -1,203 +0,0 @@
import fetch from "node-fetch";
import {
minikuraClusterRole,
minikuraClusterRoleBinding,
minikuraNamespace,
minikuraOperatorDeployment,
minikuraServiceAccount,
} from "../crds/rbac";
import type { KubernetesClient } from "./k8s-client";
import { logger } from "./logger";
export async function registerRBACResources(k8sClient: KubernetesClient): Promise<void> {
try {
logger.info("Starting RBAC resources registration");
await registerNamespace(k8sClient);
await registerServiceAccount(k8sClient);
await registerClusterRole(k8sClient);
await registerClusterRoleBinding(k8sClient);
logger.info("RBAC resources registration completed successfully");
} catch (error: any) {
logger.error(
{
err: error,
message: error.message,
statusCode: error.response?.statusCode,
body: error.response?.body,
},
"Error registering RBAC resources"
);
if (error.response) {
}
throw error;
}
}
async function registerNamespace(k8sClient: KubernetesClient): Promise<void> {
try {
const coreApi = k8sClient.getCoreApi();
await coreApi.createNamespace({ body: minikuraNamespace });
logger.info({ namespace: minikuraNamespace.metadata.name }, "Created namespace");
} catch (error: any) {
if (error.code === 409) {
logger.debug({ namespace: minikuraNamespace.metadata.name }, "Namespace already exists");
} else {
throw error;
}
}
}
async function registerServiceAccount(k8sClient: KubernetesClient): Promise<void> {
try {
const coreApi = k8sClient.getCoreApi();
await coreApi.createNamespacedServiceAccount({
namespace: minikuraServiceAccount.metadata.namespace,
body: minikuraServiceAccount,
});
logger.info(
{ serviceAccount: minikuraServiceAccount.metadata.name },
"Created service account"
);
} catch (error: any) {
if (error.code === 409) {
logger.debug(`Service account ${minikuraServiceAccount.metadata.name} already exists`);
} else {
throw error;
}
}
}
async function registerClusterRole(k8sClient: KubernetesClient): Promise<void> {
try {
const kc = k8sClient.getKubeConfig();
const opts: any = {};
await kc.applyToHTTPSOptions(opts);
const cluster = kc.getCurrentCluster();
if (!cluster) {
throw new Error("No active cluster found in KubeConfig");
}
try {
const response = await fetch(
`${cluster.server}/apis/rbac.authorization.k8s.io/v1/clusterroles`,
{
method: "POST",
headers: {
"Content-Type": "application/json",
...(opts as any).headers,
},
body: JSON.stringify(minikuraClusterRole),
agent: (opts as any).agent,
}
);
if (response.ok) {
logger.debug(`Created cluster role ${minikuraClusterRole.metadata.name}`);
} else if (response.status === 409) {
logger.debug(`Cluster role ${minikuraClusterRole.metadata.name} already exists`);
} else {
const text = await response.text();
throw new Error(
`Failed to create cluster role: ${response.status} ${response.statusText} - ${text}`
);
}
} catch (error: any) {
if (error.message?.includes("already exists") || error.message?.includes("409")) {
logger.debug(`Cluster role ${minikuraClusterRole.metadata.name} already exists`);
} else {
throw error;
}
}
} catch (error: any) {
logger.error(`Error registering cluster role:`, error.message);
throw error;
}
}
async function registerClusterRoleBinding(k8sClient: KubernetesClient): Promise<void> {
try {
const kc = k8sClient.getKubeConfig();
const opts: any = {};
await kc.applyToHTTPSOptions(opts);
const cluster = kc.getCurrentCluster();
if (!cluster) {
throw new Error("No active cluster found in KubeConfig");
}
const { default: fetch } = await import("node-fetch");
try {
const response = await fetch(
`${cluster.server}/apis/rbac.authorization.k8s.io/v1/clusterrolebindings`,
{
method: "POST",
headers: {
"Content-Type": "application/json",
...(opts as any).headers,
},
body: JSON.stringify(minikuraClusterRoleBinding),
agent: (opts as any).agent,
}
);
if (response.ok) {
logger.debug(`Created cluster role binding ${minikuraClusterRoleBinding.metadata.name}`);
} else if (response.status === 409) {
logger.debug(
`Cluster role binding ${minikuraClusterRoleBinding.metadata.name} already exists`
);
} else {
const text = await response.text();
throw new Error(
`Failed to create cluster role binding: ${response.status} ${response.statusText} - ${text}`
);
}
} catch (error: any) {
if (error.message?.includes("already exists") || error.message?.includes("409")) {
logger.debug(
`Cluster role binding ${minikuraClusterRoleBinding.metadata.name} already exists`
);
} else {
throw error;
}
}
} catch (error: any) {
logger.error(`Error registering cluster role binding:`, error.message);
throw error;
}
}
export async function registerOperatorDeployment(
k8sClient: KubernetesClient,
registryUrl: string
): Promise<void> {
try {
const deployment = JSON.parse(
JSON.stringify(minikuraOperatorDeployment).replace("${REGISTRY_URL}", registryUrl)
);
const appsApi = k8sClient.getAppsApi();
await appsApi.createNamespacedDeployment(deployment.metadata.namespace, deployment);
logger.debug(`Created deployment ${deployment.metadata.name}`);
} catch (error: any) {
if (error.code === 409) {
logger.debug(`Deployment ${minikuraOperatorDeployment.metadata.name} already exists`);
const deployment = JSON.parse(
JSON.stringify(minikuraOperatorDeployment).replace("${REGISTRY_URL}", registryUrl)
);
await k8sClient.getAppsApi().replaceNamespacedDeployment({
name: deployment.metadata.name,
namespace: deployment.metadata.namespace,
body: deployment,
});
logger.debug(`Updated deployment ${deployment.metadata.name}`);
} else {
throw error;
}
}
}
packages/k8s-operator/src/utils/service-type.ts
-19
View File
@@ -1,19 +0,0 @@
import type { ServiceType } from "@minikura/db";
export function mapServiceType(
serviceType?: ServiceType | null,
defaultType: string = "ClusterIP"
): string {
if (!serviceType) return defaultType;
switch (serviceType) {
case "CLUSTER_IP":
return "ClusterIP";
case "NODE_PORT":
return "NodePort";
case "LOAD_BALANCER":
return "LoadBalancer";
default:
return defaultType;
}
}
packages/k8s-operator/tsconfig.json
-26
View File
@@ -1,26 +0,0 @@
{
"compilerOptions": {
// Enable latest features
"lib": ["ESNext"],
"target": "ESNext",
"module": "ESNext",
"moduleDetection": "force",
"allowJs": true,
// Bundler mode
"moduleResolution": "bundler",
"allowImportingTsExtensions": true,
"verbatimModuleSyntax": true,
"noEmit": true,
// Best practices
"strict": true,
"skipLibCheck": true,
"noFallthroughCasesInSwitch": true,
// Some stricter flags (disabled by default)
"noUnusedLocals": false,
"noUnusedParameters": false,
"noPropertyAccessFromIndexSignature": false
}
}