Missing Create Namespace
This commit is contained in:
@@ -3,6 +3,7 @@ package infrastructure
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"oc-datacenter/infrastructure/minio"
|
||||
"sync"
|
||||
|
||||
@@ -41,9 +42,13 @@ func ListenNATS() {
|
||||
}
|
||||
|
||||
if argo.Type == tools.STORAGE_RESOURCE {
|
||||
fmt.Println("DETECT STORAGE ARGO_KUBE_EVENT")
|
||||
// ── Minio credential provisioning ──────────────────────────────
|
||||
setter := minio.NewMinioSetter(argo.ExecutionsID, argo.MinioID)
|
||||
if argo.SourcePeerID == argo.DestPeerID {
|
||||
fmt.Println("CONFIG MYSELF")
|
||||
err := CreateNamespace(argo.ExecutionsID)
|
||||
fmt.Println("NS", err)
|
||||
// Same peer: source creates credentials and immediately stores them.
|
||||
go setter.InitializeAsSource(context.Background(), argo.SourcePeerID, argo.DestPeerID, argo.OriginID)
|
||||
} else {
|
||||
@@ -61,6 +66,7 @@ func ListenNATS() {
|
||||
Payload: b,
|
||||
Action: tools.PB_MINIO_CONFIG,
|
||||
}); err == nil {
|
||||
fmt.Println("CONFIG THEM")
|
||||
go tools.NewNATSCaller().SetNATSPub(tools.PROPALGATION_EVENT, tools.NATSResponse{
|
||||
FromApp: "oc-datacenter",
|
||||
Datatype: -1,
|
||||
@@ -72,8 +78,12 @@ func ListenNATS() {
|
||||
}
|
||||
}
|
||||
} else {
|
||||
fmt.Println("DETECT COMPUTE ARGO_KUBE_EVENT")
|
||||
// ── Admiralty kubeconfig provisioning (existing behaviour) ──────
|
||||
if argo.SourcePeerID == argo.DestPeerID {
|
||||
fmt.Println("CONFIG MYSELF")
|
||||
err := CreateNamespace(argo.ExecutionsID)
|
||||
fmt.Println("NS", err)
|
||||
go NewAdmiraltySetter(argo.ExecutionsID).InitializeAsSource(
|
||||
context.Background(), argo.SourcePeerID, argo.DestPeerID, argo.OriginID)
|
||||
} else if b, err := json.Marshal(argo); err == nil {
|
||||
@@ -81,6 +91,7 @@ func ListenNATS() {
|
||||
Payload: b,
|
||||
Action: tools.PB_ADMIRALTY_CONFIG,
|
||||
}); err == nil {
|
||||
fmt.Println("CONFIG THEM")
|
||||
go tools.NewNATSCaller().SetNATSPub(tools.PROPALGATION_EVENT, tools.NATSResponse{
|
||||
FromApp: "oc-datacenter",
|
||||
Datatype: -1,
|
||||
@@ -93,63 +104,61 @@ func ListenNATS() {
|
||||
}
|
||||
},
|
||||
|
||||
// ─── PROPALGATION_EVENT ─────────────────────────────────────────────────────
|
||||
// Routes messages forwarded by oc-discovery to the right handler.
|
||||
tools.PROPALGATION_EVENT: func(resp tools.NATSResponse) {
|
||||
if resp.FromApp != "oc-discovery" {
|
||||
return
|
||||
}
|
||||
var prop tools.PropalgationMessage
|
||||
if err := json.Unmarshal(resp.Payload, &prop); err != nil {
|
||||
return
|
||||
}
|
||||
switch prop.Action {
|
||||
// ─── ADMIRALTY_CONFIG_EVENT ─────────────────────────────────────────────────
|
||||
// Forwarded by oc-discovery after receiving via libp2p ProtocolAdmiraltyConfigResource.
|
||||
// Payload is a KubeconfigEvent (phase discriminated by Kubeconfig presence).
|
||||
tools.ADMIRALTY_CONFIG_EVENT: func(resp tools.NATSResponse) {
|
||||
|
||||
// ── Admiralty ──────────────────────────────────────────────────────
|
||||
case tools.PB_ADMIRALTY_CONFIG:
|
||||
kubeconfigEvent := KubeconfigEvent{}
|
||||
if err := json.Unmarshal(prop.Payload, &kubeconfigEvent); err == nil {
|
||||
if kubeconfigEvent.Kubeconfig != "" {
|
||||
// Phase 2: kubeconfig present → this peer is the TARGET (scheduler).
|
||||
NewAdmiraltySetter(kubeconfigEvent.ExecutionsID).InitializeAsTarget(
|
||||
context.Background(), kubeconfigEvent)
|
||||
} else {
|
||||
// Phase 1: no kubeconfig → this peer is the SOURCE (compute).
|
||||
NewAdmiraltySetter(kubeconfigEvent.ExecutionsID).InitializeAsSource(
|
||||
context.Background(), kubeconfigEvent.SourcePeerID, kubeconfigEvent.DestPeerID, kubeconfigEvent.OriginID)
|
||||
}
|
||||
}
|
||||
|
||||
// ── Minio ──────────────────────────────────────────────────────────
|
||||
case tools.PB_MINIO_CONFIG:
|
||||
minioEvent := minio.MinioCredentialEvent{}
|
||||
if err := json.Unmarshal(prop.Payload, &minioEvent); err == nil {
|
||||
if minioEvent.Access != "" {
|
||||
// Phase 2: credentials present → this peer is the TARGET (compute).
|
||||
minio.NewMinioSetter(minioEvent.ExecutionsID, minioEvent.MinioID).InitializeAsTarget(
|
||||
context.Background(), minioEvent)
|
||||
} else {
|
||||
// Phase 1: no credentials → this peer is the SOURCE (Minio host).
|
||||
minio.NewMinioSetter(minioEvent.ExecutionsID, minioEvent.MinioID).InitializeAsSource(
|
||||
context.Background(), minioEvent.SourcePeerID, minioEvent.DestPeerID, minioEvent.OriginID)
|
||||
}
|
||||
}
|
||||
|
||||
// ── Deletion (routed by oc-discovery to the source peer) ───────────
|
||||
case tools.PB_DELETE:
|
||||
argo := &ArgoKubeEvent{}
|
||||
if err := json.Unmarshal(prop.Payload, argo); err != nil || argo.ExecutionsID == "" {
|
||||
return
|
||||
}
|
||||
if argo.Type == tools.STORAGE_RESOURCE {
|
||||
// Minio source teardown: revoke credentials + delete bucket.
|
||||
deleteEvent := minio.MinioDeleteEvent{}
|
||||
if err := json.Unmarshal(prop.Payload, &deleteEvent); err == nil {
|
||||
go minio.NewMinioSetter(deleteEvent.ExecutionsID, deleteEvent.MinioID).
|
||||
TeardownAsSource(context.Background(), deleteEvent)
|
||||
}
|
||||
kubeconfigEvent := KubeconfigEvent{}
|
||||
if err := json.Unmarshal(resp.Payload, &kubeconfigEvent); err == nil {
|
||||
if kubeconfigEvent.Kubeconfig != "" {
|
||||
// Phase 2: kubeconfig present → this peer is the TARGET (scheduler).
|
||||
NewAdmiraltySetter(kubeconfigEvent.ExecutionsID).InitializeAsTarget(
|
||||
context.Background(), kubeconfigEvent)
|
||||
} else {
|
||||
// Admiralty source teardown: delete AdmiraltySource + namespace.
|
||||
err := CreateNamespace(kubeconfigEvent.ExecutionsID)
|
||||
fmt.Println("NS", err)
|
||||
// Phase 1: no kubeconfig → this peer is the SOURCE (compute).
|
||||
NewAdmiraltySetter(kubeconfigEvent.ExecutionsID).InitializeAsSource(
|
||||
context.Background(), kubeconfigEvent.SourcePeerID, kubeconfigEvent.DestPeerID, kubeconfigEvent.OriginID)
|
||||
}
|
||||
}
|
||||
},
|
||||
|
||||
// ─── MINIO_CONFIG_EVENT ──────────────────────────────────────────────────────
|
||||
// Forwarded by oc-discovery after receiving via libp2p ProtocolMinioConfigResource.
|
||||
// Payload is a MinioCredentialEvent (phase discriminated by Access presence).
|
||||
tools.MINIO_CONFIG_EVENT: func(resp tools.NATSResponse) {
|
||||
minioEvent := minio.MinioCredentialEvent{}
|
||||
if err := json.Unmarshal(resp.Payload, &minioEvent); err == nil {
|
||||
if minioEvent.Access != "" {
|
||||
// Phase 2: credentials present → this peer is the TARGET (compute).
|
||||
minio.NewMinioSetter(minioEvent.ExecutionsID, minioEvent.MinioID).InitializeAsTarget(
|
||||
context.Background(), minioEvent)
|
||||
} else {
|
||||
err := CreateNamespace(minioEvent.ExecutionsID)
|
||||
fmt.Println("NS", err)
|
||||
// Phase 1: no credentials → this peer is the SOURCE (Minio host).
|
||||
minio.NewMinioSetter(minioEvent.ExecutionsID, minioEvent.MinioID).InitializeAsSource(
|
||||
context.Background(), minioEvent.SourcePeerID, minioEvent.DestPeerID, minioEvent.OriginID)
|
||||
}
|
||||
}
|
||||
},
|
||||
|
||||
// ─── REMOVE_RESOURCE ────────────────────────────────────────────────────────
|
||||
// Routed by oc-discovery via ProtocolDeleteResource for datacenter teardown.
|
||||
// Only STORAGE_RESOURCE and COMPUTE_RESOURCE deletions are handled here.
|
||||
tools.REMOVE_RESOURCE: func(resp tools.NATSResponse) {
|
||||
switch resp.Datatype {
|
||||
case tools.STORAGE_RESOURCE:
|
||||
deleteEvent := minio.MinioDeleteEvent{}
|
||||
if err := json.Unmarshal(resp.Payload, &deleteEvent); err == nil && deleteEvent.ExecutionsID != "" {
|
||||
go minio.NewMinioSetter(deleteEvent.ExecutionsID, deleteEvent.MinioID).
|
||||
TeardownAsSource(context.Background(), deleteEvent)
|
||||
}
|
||||
case tools.COMPUTE_RESOURCE:
|
||||
argo := &ArgoKubeEvent{}
|
||||
if err := json.Unmarshal(resp.Payload, argo); err == nil && argo.ExecutionsID != "" {
|
||||
go NewAdmiraltySetter(argo.ExecutionsID).TeardownAsSource(context.Background())
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user