Some logging added.
This change is untested!
This commit is contained in:
@@ -3,6 +3,7 @@ package config;
|
|||||||
import (
|
import (
|
||||||
"os"
|
"os"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
|
"errors"
|
||||||
);
|
);
|
||||||
|
|
||||||
type Config struct {
|
type Config struct {
|
||||||
@@ -26,14 +27,15 @@ type RegistryAuth struct {
|
|||||||
Password string
|
Password string
|
||||||
}
|
}
|
||||||
|
|
||||||
func GetConfig() Config {
|
func GetConfig()(Config, error){
|
||||||
var cfg Config;
|
var cfg Config;
|
||||||
|
|
||||||
rawConfig, err := os.ReadFile("/config.json");
|
rawConfig, err := os.ReadFile("/config.json");
|
||||||
if err != nil{
|
if err != nil{
|
||||||
panic("Failed it load config file." + err.Error());
|
return cfg, errors.New("Failed it load config file." + err.Error());
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Set defaults
|
||||||
cfg.Constants.OverlayNetworkName = "blazenaPohar";
|
cfg.Constants.OverlayNetworkName = "blazenaPohar";
|
||||||
cfg.Constants.HelperServiceName = "blazenaHelper";
|
cfg.Constants.HelperServiceName = "blazenaHelper";
|
||||||
cfg.Constants.StorageContainerName = "blazenaStorage";
|
cfg.Constants.StorageContainerName = "blazenaStorage";
|
||||||
@@ -42,8 +44,8 @@ func GetConfig() Config {
|
|||||||
err = json.Unmarshal(rawConfig, &cfg);
|
err = json.Unmarshal(rawConfig, &cfg);
|
||||||
|
|
||||||
if err != nil{
|
if err != nil{
|
||||||
panic("Failed to unmarshal config." + err.Error())
|
return cfg, errors.New("Failed to unmarshal config: " + err.Error());
|
||||||
}
|
}
|
||||||
|
|
||||||
return cfg;
|
return cfg, err;
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -4,6 +4,7 @@ import (
|
|||||||
"context"
|
"context"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
"fmt"
|
"fmt"
|
||||||
|
"log/slog"
|
||||||
"os"
|
"os"
|
||||||
"os/signal"
|
"os/signal"
|
||||||
"strings"
|
"strings"
|
||||||
@@ -32,22 +33,27 @@ type aService struct{
|
|||||||
}
|
}
|
||||||
|
|
||||||
func Run(Config cfg.Config){
|
func Run(Config cfg.Config){
|
||||||
|
// Before touching the line below think.
|
||||||
theConfig = Config;
|
theConfig = Config;
|
||||||
|
|
||||||
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM);
|
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM);
|
||||||
|
|
||||||
var err error;
|
var err error;
|
||||||
ApiClient, err = client.NewClientWithOpts(client.FromEnv);
|
ApiClient, err = client.NewClientWithOpts(client.FromEnv);
|
||||||
if(err != nil){
|
if(err != nil){
|
||||||
panic("Docker client was not able to init from env!" + err.Error());
|
slog.Error("Failed to create docker client!", slog.String("note", "Try to look into DOCKER_HOST env var or check if socket exists and works"));
|
||||||
|
os.Exit(1);
|
||||||
}
|
}
|
||||||
|
|
||||||
info, err := ApiClient.Info(context.Background())
|
info, err := ApiClient.Info(context.Background())
|
||||||
if(err != nil){
|
if(err != nil){
|
||||||
panic("Error getting info!" + err.Error());
|
slog.Error("Error getting info from docker socket!", slog.String("note", "This is kind of ping."));
|
||||||
|
os.Exit(1);
|
||||||
}
|
}
|
||||||
|
|
||||||
if(!info.Swarm.ControlAvailable){
|
if(!info.Swarm.ControlAvailable){
|
||||||
panic("Node is not a swarm manager.");
|
slog.Error("This node is not a swarm manager!");
|
||||||
|
os.Exit(1);
|
||||||
}
|
}
|
||||||
|
|
||||||
server := &http.Server{
|
server := &http.Server{
|
||||||
@@ -68,7 +74,7 @@ func Run(Config cfg.Config){
|
|||||||
stop();
|
stop();
|
||||||
});
|
});
|
||||||
|
|
||||||
ApiClient.NetworkCreate(context.Background(), theConfig.Constants.OverlayNetworkName, network.CreateOptions{
|
ApiClient.NetworkCreate(context.Background(), Config.Constants.OverlayNetworkName, network.CreateOptions{
|
||||||
Attachable: true,
|
Attachable: true,
|
||||||
// Internal: true,
|
// Internal: true,
|
||||||
Driver: "overlay",
|
Driver: "overlay",
|
||||||
@@ -84,7 +90,8 @@ func Run(Config cfg.Config){
|
|||||||
|
|
||||||
}
|
}
|
||||||
if(err != nil){
|
if(err != nil){
|
||||||
panic("Unable to start http server!" + err.Error());
|
slog.Error("Unable to start http server!", slog.Any("propagatedError", err));
|
||||||
|
os.Exit(1);
|
||||||
}
|
}
|
||||||
|
|
||||||
}();
|
}();
|
||||||
@@ -96,7 +103,7 @@ func Run(Config cfg.Config){
|
|||||||
fmt.Println("Stopping http server.");
|
fmt.Println("Stopping http server.");
|
||||||
server.Close();
|
server.Close();
|
||||||
|
|
||||||
ApiClient.NetworkRemove(context.Background(), theConfig.Constants.OverlayNetworkName);
|
ApiClient.NetworkRemove(context.Background(), Config.Constants.OverlayNetworkName);
|
||||||
ApiClient.ConfigRemove(context.Background(), "blazenaSSHPublicKey")
|
ApiClient.ConfigRemove(context.Background(), "blazenaSSHPublicKey")
|
||||||
ApiClient.SecretRemove(context.Background(), "blazenaSSHHostPrivateKey");
|
ApiClient.SecretRemove(context.Background(), "blazenaSSHHostPrivateKey");
|
||||||
|
|
||||||
@@ -110,6 +117,7 @@ func bearerAuth(w http.ResponseWriter, r *http.Request)bool {
|
|||||||
if authHeader != expected {
|
if authHeader != expected {
|
||||||
w.WriteHeader(http.StatusUnauthorized)
|
w.WriteHeader(http.StatusUnauthorized)
|
||||||
fmt.Fprintln(w, "Unauthorized")
|
fmt.Fprintln(w, "Unauthorized")
|
||||||
|
slog.Warn("Unauthorized request received", slog.Any("request", *r));
|
||||||
return false;
|
return false;
|
||||||
}
|
}
|
||||||
return true;
|
return true;
|
||||||
@@ -131,6 +139,16 @@ func listServices(w http.ResponseWriter, r *http.Request){
|
|||||||
|
|
||||||
var services []aService;
|
var services []aService;
|
||||||
|
|
||||||
|
|
||||||
|
nodes, err := ApiClient.NodeList(context.Background(), swarm.NodeListOptions{});
|
||||||
|
|
||||||
|
var validNodeHostnames []string;
|
||||||
|
|
||||||
|
for _, node := range nodes{
|
||||||
|
validNodeHostnames = append(validNodeHostnames, node.Description.Hostname);
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
for _, service := range list{
|
for _, service := range list{
|
||||||
var settings map[string]string = service.Spec.Labels;
|
var settings map[string]string = service.Spec.Labels;
|
||||||
|
|
||||||
@@ -141,6 +159,12 @@ func listServices(w http.ResponseWriter, r *http.Request){
|
|||||||
|
|
||||||
targetVolumes := strings.Split(settings["blazena.volumes"], ",");
|
targetVolumes := strings.Split(settings["blazena.volumes"], ",");
|
||||||
|
|
||||||
|
if !contains(validNodeHostnames, settings["blazena.node"]) {
|
||||||
|
errMsg := "node with hostname:'"+ settings["blazena.node"] +"' does not exist.";
|
||||||
|
slog.Warn("Invalid Service Config!", slog.String("serviceId", service.ID), slog.String("errMessage", errMsg));
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
|
||||||
services = append(services, aService{
|
services = append(services, aService{
|
||||||
ServiceId: service.ID,
|
ServiceId: service.ID,
|
||||||
VolumeNames: targetVolumes,
|
VolumeNames: targetVolumes,
|
||||||
@@ -152,7 +176,8 @@ func listServices(w http.ResponseWriter, r *http.Request){
|
|||||||
bytes, err := json.Marshal(services);
|
bytes, err := json.Marshal(services);
|
||||||
|
|
||||||
if err != nil{
|
if err != nil{
|
||||||
panic("Error during response encoding!" + err.Error());
|
slog.Error("Error during response encoding!", slog.Any("propagatedError", err));
|
||||||
|
os.Exit(1);
|
||||||
}
|
}
|
||||||
|
|
||||||
fmt.Fprint(w, string(bytes));
|
fmt.Fprint(w, string(bytes));
|
||||||
|
|||||||
@@ -2,17 +2,17 @@ package docker
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
|
"encoding/base64"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
"fmt"
|
"fmt"
|
||||||
"io"
|
"io"
|
||||||
"net/http"
|
"net/http"
|
||||||
"encoding/base64"
|
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"github.com/docker/docker/api/types/mount"
|
|
||||||
"github.com/docker/docker/api/types/swarm"
|
|
||||||
"github.com/docker/docker/api/types/registry"
|
|
||||||
"github.com/docker/docker/api/types/image"
|
"github.com/docker/docker/api/types/image"
|
||||||
|
"github.com/docker/docker/api/types/mount"
|
||||||
|
"github.com/docker/docker/api/types/registry"
|
||||||
|
"github.com/docker/docker/api/types/swarm"
|
||||||
"github.com/docker/docker/client"
|
"github.com/docker/docker/client"
|
||||||
|
|
||||||
cfg "github.com/rony5394/blazena/config"
|
cfg "github.com/rony5394/blazena/config"
|
||||||
|
|||||||
1
go.mod
1
go.mod
@@ -19,6 +19,7 @@ require (
|
|||||||
github.com/felixge/httpsnoop v1.0.4 // indirect
|
github.com/felixge/httpsnoop v1.0.4 // indirect
|
||||||
github.com/go-logr/logr v1.4.3 // indirect
|
github.com/go-logr/logr v1.4.3 // indirect
|
||||||
github.com/go-logr/stdr v1.2.2 // indirect
|
github.com/go-logr/stdr v1.2.2 // indirect
|
||||||
|
github.com/google/uuid v1.6.0 // indirect
|
||||||
github.com/moby/docker-image-spec v1.3.1 // indirect
|
github.com/moby/docker-image-spec v1.3.1 // indirect
|
||||||
github.com/moby/sys/atomicwriter v0.1.0 // indirect
|
github.com/moby/sys/atomicwriter v0.1.0 // indirect
|
||||||
github.com/moby/term v0.5.2 // indirect
|
github.com/moby/term v0.5.2 // indirect
|
||||||
|
|||||||
46
host/host.go
46
host/host.go
@@ -8,6 +8,7 @@ import (
|
|||||||
"errors"
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
"io"
|
"io"
|
||||||
|
"log/slog"
|
||||||
"net/http"
|
"net/http"
|
||||||
"os"
|
"os"
|
||||||
"time"
|
"time"
|
||||||
@@ -44,18 +45,24 @@ func Run(Config cfg.Config) {
|
|||||||
services := getServices(Config);
|
services := getServices(Config);
|
||||||
|
|
||||||
for _, service := range services {
|
for _, service := range services {
|
||||||
fmt.Println("Scaling Down: "+service.ServiceId)
|
|
||||||
scale(Config, service.ServiceId, false);
|
|
||||||
fmt.Println("Done!");
|
|
||||||
for _, volume := range service.VolumeNames{
|
|
||||||
fmt.Println("Preparing: " + service.ServiceId + " volume: " + volume);
|
|
||||||
if !prepareService(Config, service, volume) {continue}
|
|
||||||
fmt.Println("Done!");
|
|
||||||
|
|
||||||
storagePath, _ := generateStoragePath(Config, service.Node, volume, DockerClient);
|
slog.Info("Scaling Down", slog.String("serviceId", service.ServiceId));
|
||||||
fmt.Println(storagePath);
|
scale(Config, service.ServiceId, false);
|
||||||
|
slog.Info("Done");
|
||||||
|
|
||||||
|
for _, volume := range service.VolumeNames{
|
||||||
|
slog.Info("Preparing", slog.String("serviceId", service.ServiceId), slog.String("volumeId", volume));
|
||||||
|
if !prepareService(Config, service, volume) {continue}
|
||||||
|
slog.Info("Done");
|
||||||
|
|
||||||
|
targetStoragePath, _ := generateStoragePath(Config, service.Node, volume, DockerClient);
|
||||||
|
sourceStoragePath := "root@tasks."+ Config.Constants.HelperServiceName +":/volume";
|
||||||
|
|
||||||
|
slog.Debug("targetStoragePath", slog.String("value", targetStoragePath), slog.String("serviceId", service.ServiceId));
|
||||||
|
slog.Debug("sourceStoragePath", slog.String("value", sourceStoragePath), slog.String("serviceId", service.ServiceId));
|
||||||
|
|
||||||
command := `rsync -avz --delete -e "ssh -i /ssh-key -p 2222 -o StrictHostKeyChecking=yes -o UserKnownHostsFile=/expected-host-key" \
|
command := `rsync -avz --delete -e "ssh -i /ssh-key -p 2222 -o StrictHostKeyChecking=yes -o UserKnownHostsFile=/expected-host-key" \
|
||||||
root@tasks.`+ Config.Constants.HelperServiceName +`:/volume/ ` +storagePath;
|
`+ sourceStoragePath +" "+ targetStoragePath;
|
||||||
|
|
||||||
exec, err := DockerClient.ContainerExecCreate(context.Background(), Config.Constants.StorageContainerName, container.ExecOptions{
|
exec, err := DockerClient.ContainerExecCreate(context.Background(), Config.Constants.StorageContainerName, container.ExecOptions{
|
||||||
Cmd: []string{"sh", "-c", command},
|
Cmd: []string{"sh", "-c", command},
|
||||||
@@ -64,30 +71,37 @@ func Run(Config cfg.Config) {
|
|||||||
Tty: false,
|
Tty: false,
|
||||||
});
|
});
|
||||||
if err != nil {
|
if err != nil {
|
||||||
panic("Failed to create rsync exec!"+err.Error());
|
slog.Error("Failed to create rsync exec!", slog.Any("propagatedError", err));
|
||||||
|
os.Exit(1);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
resp, err := DockerClient.ContainerExecAttach(context.Background(), exec.ID, container.ExecStartOptions{});
|
resp, err := DockerClient.ContainerExecAttach(context.Background(), exec.ID, container.ExecStartOptions{});
|
||||||
|
if err != nil {
|
||||||
|
slog.Error("Failed to create container exec!", slog.Any("propagatedError", err));
|
||||||
|
}
|
||||||
defer resp.Close();
|
defer resp.Close();
|
||||||
|
|
||||||
io.Copy(os.Stdout, resp.Reader)
|
io.Copy(os.Stdout, resp.Reader)
|
||||||
|
|
||||||
time.Sleep(30*time.Second);
|
time.Sleep(30*time.Second);
|
||||||
fmt.Println("Cleaning Up: " + service.ServiceId);
|
slog.Info("Cleaning Up", slog.String("serviceId", service.ServiceId), slog.String("volumeId", volume));
|
||||||
cleanupService(Config, service);
|
cleanupService(Config, service);
|
||||||
fmt.Println("Done!");
|
slog.Info("Done!");
|
||||||
}
|
}
|
||||||
fmt.Println("Scaling up: "+service.ServiceId);
|
slog.Info("Scaling Up", slog.String("serviceId", service.ServiceId));
|
||||||
scale(Config, service.ServiceId, true);
|
scale(Config, service.ServiceId, true);
|
||||||
fmt.Println("Done!");
|
slog.Info("Done!");
|
||||||
}
|
}
|
||||||
|
|
||||||
DockerClient.ContainerRemove(context.Background(), Config.Constants.StorageContainerName, container.RemoveOptions{
|
DockerClient.ContainerRemove(context.Background(), Config.Constants.StorageContainerName, container.RemoveOptions{
|
||||||
Force: true,
|
Force: true,
|
||||||
});
|
});
|
||||||
|
|
||||||
if !shutdown(Config){panic("Failed to shutdown docker api!");}
|
if !shutdown(Config){
|
||||||
|
slog.Error("Failed to shutdown docker api!");
|
||||||
|
os.Exit(1);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func getServices(Config cfg.Config)[]aService{
|
func getServices(Config cfg.Config)[]aService{
|
||||||
|
|||||||
15
main.go
15
main.go
@@ -5,6 +5,7 @@ import (
|
|||||||
"github.com/rony5394/blazena/docker"
|
"github.com/rony5394/blazena/docker"
|
||||||
"github.com/rony5394/blazena/host"
|
"github.com/rony5394/blazena/host"
|
||||||
cfg "github.com/rony5394/blazena/config"
|
cfg "github.com/rony5394/blazena/config"
|
||||||
|
"log/slog"
|
||||||
);
|
);
|
||||||
|
|
||||||
func main() {
|
func main() {
|
||||||
@@ -12,7 +13,19 @@ func main() {
|
|||||||
panic("Usage: blazena <mode>");
|
panic("Usage: blazena <mode>");
|
||||||
}
|
}
|
||||||
|
|
||||||
var config = cfg.GetConfig();
|
logger := slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{
|
||||||
|
Level: slog.LevelDebug,
|
||||||
|
}));
|
||||||
|
slog.SetDefault(logger);
|
||||||
|
|
||||||
|
config, err := cfg.GetConfig();
|
||||||
|
|
||||||
|
if(err != nil){
|
||||||
|
slog.Error("Failed to load config!", slog.Any("propagatedError", err.Error()));
|
||||||
|
os.Exit(1);
|
||||||
|
}
|
||||||
|
|
||||||
|
slog.Debug("Config", slog.Any("Value", config));
|
||||||
|
|
||||||
mode := os.Args[1];
|
mode := os.Args[1];
|
||||||
switch mode {
|
switch mode {
|
||||||
|
|||||||
28
shared/trace.go
Normal file
28
shared/trace.go
Normal file
@@ -0,0 +1,28 @@
|
|||||||
|
package shared
|
||||||
|
|
||||||
|
import (
|
||||||
|
"log/slog"
|
||||||
|
|
||||||
|
"github.com/google/uuid"
|
||||||
|
);
|
||||||
|
|
||||||
|
func NewTraceId()string{
|
||||||
|
return uuid.New().String();
|
||||||
|
}
|
||||||
|
|
||||||
|
func helper(name string, id *string)slog.Attr{
|
||||||
|
if id == nil{
|
||||||
|
return slog.String(name, NewTraceId());
|
||||||
|
}
|
||||||
|
|
||||||
|
return slog.String(name, *id);
|
||||||
|
}
|
||||||
|
|
||||||
|
func NewSlogTrace(id *string)slog.Attr{
|
||||||
|
return helper("trace", id);
|
||||||
|
}
|
||||||
|
|
||||||
|
func NewSlogOperation(id *string)slog.Attr{
|
||||||
|
return helper("operation", id);
|
||||||
|
}
|
||||||
|
|
||||||
Reference in New Issue
Block a user