rony539/blazena
1
0
Fork 0
Code Issues Pull Requests Activity

Something

Browse Source
This commit is contained in:
rony5394
2026-02-20 21:47:11 +01:00
parent 91c16cf429
commit 11f01ab459
5 changed files with 174 additions and 90 deletions
Download Patch File Download Diff File Expand all files Collapse all files

62
docker/cleanup.go Normal file
View File

@@ -0,0 +1,62 @@
package docker
import (
"context"
"encoding/json"
"fmt"
"io"
"net/http"
"github.com/moby/moby/client"
)
func cleanup(w http.ResponseWriter, r *http.Request){
if r.Method != http.MethodPost{
w.WriteHeader(http.StatusMethodNotAllowed);
fmt.Fprint(w, "Method Not Allowed");
return;
}
if !bearerAuth(w, r) {return;}
rawBody, err := io.ReadAll(r.Body);
if err != nil {
panic("Failed to read body!");
}
var bodyDecoded struct{
ServiceId string `json:"serviceId"`
};
err = json.Unmarshal(rawBody, &bodyDecoded);
if err != nil {
panic("Failed to unmarshal json."+ err.Error());
}
scaleUp(bodyDecoded.ServiceId);
listResoult, err := ApiClient.ServiceList(context.Background(), client.ServiceListOptions{});
if err != nil {
panic("Failed to list services."+ err.Error());
}
var helperServiceId string;
for _, service := range listResoult.Items {
if service.Spec.Labels["blazena.helper"] != "true" {
continue;
}
helperServiceId = service.ID;
break;
}
if helperServiceId == ""{
panic("Helper service not found!");
}
_, err = ApiClient.ServiceRemove(context.Background(), helperServiceId, client.ServiceRemoveOptions{});
if err != nil {
panic("Failed to remove helper service."+ err.Error());
}
fmt.Fprint(w, bodyDecoded.ServiceId);
}

4
docker/docker.go
View File

@@ -50,9 +50,8 @@ func Run(){
}
http.HandleFunc("/services", listServices);
http.HandleFunc("/scale/down", scaleDown);
http.HandleFunc("/scale/up", scaleUp);
http.HandleFunc("/prepare", prepare);
http.HandleFunc("/cleanup", cleanup);
go func(){
err = server.ListenAndServe();
if err == http.ErrServerClosed {
@@ -87,6 +86,7 @@ func bearerAuth(w http.ResponseWriter, r *http.Request)bool {
func listServices(w http.ResponseWriter, r *http.Request){
if(r.Method != http.MethodGet){
w.WriteHeader(http.StatusMethodNotAllowed);
fmt.Fprintln(w, "Method Not Allowed");
return;
}

93
docker/prepare.go
View File

@@ -6,9 +6,12 @@ import (
"fmt"
"io"
"net/http"
"strings"
"time"
"github.com/moby/moby/api/types/swarm"
"github.com/moby/moby/api/types/mount"
"github.com/moby/moby/api/types/network"
"github.com/moby/moby/api/types/swarm"
"github.com/moby/moby/client"
)
@@ -19,7 +22,9 @@ func prepare(w http.ResponseWriter, r *http.Request){
return;
}
//TODO: add token auth
//if !bearerAuth(w, r){
// return;
//}
rawBody, err := io.ReadAll(r.Body);
if err != nil {
@@ -27,8 +32,7 @@ func prepare(w http.ResponseWriter, r *http.Request){
}
var bodyDecoded struct{
volume string
node string
ServiceId string `json:"serviceId"`
};
err = json.Unmarshal(rawBody, &bodyDecoded);
@@ -36,34 +40,71 @@ func prepare(w http.ResponseWriter, r *http.Request){
panic("Failed to unmarshal json."+ err.Error());
}
scaleDown(bodyDecoded.ServiceId);
inspectResoults, err := ApiClient.ServiceInspect(context.Background(), bodyDecoded.ServiceId, client.ServiceInspectOptions{});
if err != nil{
panic("Failed to inspect service."+ err.Error());
}
labels := inspectResoults.Service.Spec.Labels;
time.Sleep(10);
maxConcurrent := uint64(1);
totalCompletions := uint64(1);
targetVolume := bodyDecoded.volume;
targetNode := bodyDecoded.node;
targetVolumes := strings.Split(labels["blazena.volumes"], ",");
targetNode := labels["blazena.node"];
helperCommand := `apk add openssh rsync && \
ssh-keygen -t ed25519 -f /host_key && \
mkdir -p /root/.ssh/ && \
echo "ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIByYbl8vu946LPycSO5pBohq3vMvvl+wX7snu1Bqpd7p test" > /root/.ssh/authorized_keys && \
/usr/sbin/sshd -h /host_key -p 22 -D`;
ApiClient.ServiceCreate(context.Background(), client.ServiceCreateOptions{
Spec: swarm.ServiceSpec{
Mode: swarm.ServiceMode{
ReplicatedJob: &swarm.ReplicatedJob{
MaxConcurrent: &maxConcurrent,
TotalCompletions: &totalCompletions,
for _, targetVolume := range targetVolumes{
_, err := ApiClient.ServiceCreate(context.Background(), client.ServiceCreateOptions{
Spec: swarm.ServiceSpec{
Annotations: swarm.Annotations{
Labels: map[string]string{"blazena.helper": "true"},
},
},
TaskTemplate: swarm.TaskSpec{
ContainerSpec: &swarm.ContainerSpec{
Image: "docker.io/library/alpine:latest",
Mounts: []mount.Mount{
mount.Mount{
Source: targetVolume,
Target: "/volume",
Type: "bind",
Mode: swarm.ServiceMode{
ReplicatedJob: &swarm.ReplicatedJob{
MaxConcurrent: &maxConcurrent,
TotalCompletions: &totalCompletions,
},
},
TaskTemplate: swarm.TaskSpec{
ContainerSpec: &swarm.ContainerSpec{
Image: "docker.io/library/alpine:latest",
Command: []string{"sh", "-c", helperCommand},
Mounts: []mount.Mount{
mount.Mount{
Source: targetVolume,
Target: "/volume",
Type: "volume",
},
},
},
Placement: &swarm.Placement{
Constraints: []string{"node.hostname=="+targetNode},
},
},
EndpointSpec: &swarm.EndpointSpec{
Ports: []swarm.PortConfig{
swarm.PortConfig{
Protocol: network.TCP,
TargetPort: uint32(22),
PublishedPort: uint32(2222),
},
},
},
Placement: &swarm.Placement{
Constraints: []string{"node.hostname=="+targetNode},
},
},
},
});
});
if err != nil {
panic("Failed to create helper service."+ err.Error());
}
fmt.Fprint(w, bodyDecoded.ServiceId);
}
}

68
docker/scale.go
View File

@@ -1,35 +1,13 @@
package docker;
import(
"net/http"
"io"
"encoding/json"
package docker
import (
"context"
"github.com/moby/moby/client"
);
func scaleDown(w http.ResponseWriter, r *http.Request){
if r.Method != http.MethodPost {
w.WriteHeader(http.StatusMethodNotAllowed);
return;
}
if !bearerAuth(w, r) {return}
rawBody, err := io.ReadAll(r.Body);
if err != nil {
panic("Failed to read body!" + err.Error());
}
var parsedBody struct{
ServiceId string `json:"serviceId"`;
};
err = json.Unmarshal(rawBody, &parsedBody);
if err != nil{
panic("Failed to unmarshal request body!"+ err.Error());
}
inspectresoult, err := ApiClient.ServiceInspect(context.Background(), parsedBody.ServiceId, client.ServiceInspectOptions{});
func scaleDown(serviceId string){
inspectresoult, err := ApiClient.ServiceInspect(context.Background(), serviceId, client.ServiceInspectOptions{});
if err != nil{
panic("Error inspecting service!"+ err.Error());
@@ -41,9 +19,9 @@ func scaleDown(w http.ResponseWriter, r *http.Request){
newScale := uint64(0);
updatedSpec.Mode.Replicated.Replicas = &newScale;
scale.Store(parsedBody.ServiceId, *originalScale);
scale.Store(serviceId, *originalScale);
_, err = ApiClient.ServiceUpdate(context.Background(), parsedBody.ServiceId, client.ServiceUpdateOptions{
_, err = ApiClient.ServiceUpdate(context.Background(), serviceId, client.ServiceUpdateOptions{
Spec: updatedSpec,
Version: inspectresoult.Service.Version,
});
@@ -53,36 +31,14 @@ func scaleDown(w http.ResponseWriter, r *http.Request){
}
}
func scaleUp(w http.ResponseWriter, r *http.Request){
if r.Method != http.MethodPost {
w.WriteHeader(http.StatusMethodNotAllowed);
return;
}
if !bearerAuth(w, r) {return}
rawBody, err := io.ReadAll(r.Body);
if err != nil {
panic("Failed to read body!");
}
var parsedBody struct{
ServiceId string `json:"serviceId"`;
};
err = json.Unmarshal(rawBody, &parsedBody);
if err != nil{
panic("Failed to unmarshal request body!"+ err.Error());
}
inspectresoult, err := ApiClient.ServiceInspect(context.Background(), parsedBody.ServiceId, client.ServiceInspectOptions{});
func scaleUp(serviceId string){
inspectresoult, err := ApiClient.ServiceInspect(context.Background(), serviceId, client.ServiceInspectOptions{});
if err != nil{
panic("Error inspecting service!"+ err.Error());
}
originalScale, ok := scale.Load(parsedBody.ServiceId);
originalScale, ok := scale.Load(serviceId);
if(!ok){
panic("Its not okay!");
}
@@ -96,7 +52,7 @@ func scaleUp(w http.ResponseWriter, r *http.Request){
updatedSpec.Mode.Replicated.Replicas = &originalScaleChecked;
ApiClient.ServiceUpdate(context.Background(), parsedBody.ServiceId, client.ServiceUpdateOptions{
ApiClient.ServiceUpdate(context.Background(), serviceId, client.ServiceUpdateOptions{
Spec: updatedSpec,
Version: inspectresoult.Service.Version,

37
host/host.go
View File

@@ -1,15 +1,19 @@
package host
import (
"bytes"
"encoding/json"
"fmt"
"io"
"net/http"
"os"
cfg "github.com/rony5394/blazena/config"
"github.com/moby/moby/client"
)
var token string = "12345";
var DockerClient *client.Client;
type aService struct{
ServiceId string `json:"serviceId"`;
@@ -26,15 +30,36 @@ func Run(Config cfg.Config) {
fmt.Println("Node", service.Node, "refferenced in", service.ServiceId ,"service does not exists!");
continue;
}
for _, volume := range service.VolumeNames{
remoteFilePath := Config.Nodes[service.Node].DockerVolumePath + "/" + volume;
remotePath := Config.User + "@" + Config.Nodes[service.Node].Ip + ":" + remoteFilePath;
base := "rsync -avh --delete --progress -e 'ssh -i ./ssh-key' ";
localPath := Config.LocalBasePath + "/@current/@" + service.Node;
fmt.Println(base + remotePath + " " + localPath);
var body struct{
ServiceId string `json:"serviceId"`
} = struct{ServiceId string "json:\"serviceId\""}{
ServiceId: service.ServiceId,
}
bodyEncoded, err := json.Marshal(body);
if err != nil {
panic("Failed to marshal body."+ err.Error());
}
rq, err := http.NewRequest("POST", Config.DockerManagerBaseUrl + "/prepare", bytes.NewBuffer(bodyEncoded));
if err != nil{
panic("Failed to create http request"+ err.Error());
}
rq.Header.Set("Authorization", "Bearer "+ token);
rq.Close = true;
rs, err := http.DefaultClient.Do(rq);
defer rs.Body.Close();
if err != nil{
panic("Failed to send http request"+ err.Error());
}
os.Exit(0);
}
}