small bug fixes and organization

This commit is contained in:
Zoe
2024-12-09 06:03:45 -06:00
parent 54fbb02ad8
commit 6c035fc391
12 changed files with 289 additions and 289 deletions

239
server/app.go Normal file
View File

@@ -0,0 +1,239 @@
package server
import (
"context"
"database/sql"
"fmt"
"log"
"os"
"path/filepath"
"sync"
"github.com/juls0730/flux/pkg"
)
type App struct {
ID int64 `json:"id,omitempty"`
Deployment Deployment `json:"-"`
Name string `json:"name,omitempty"`
DeploymentID int64 `json:"deployment_id,omitempty"`
}
func CreateApp(ctx context.Context, imageName string, projectPath string, projectConfig pkg.ProjectConfig) (*App, error) {
app := &App{
Name: projectConfig.Name,
}
log.Printf("Creating deployment %s...\n", app.Name)
container, err := CreateDockerContainer(ctx, imageName, projectPath, projectConfig)
if err != nil || container == nil {
return nil, fmt.Errorf("Failed to create container: %v", err)
}
deployment, err := CreateDeployment(*container, projectConfig.Port, projectConfig.Url, Flux.db)
app.Deployment = deployment
if err != nil {
log.Printf("Failed to create deployment: %v", err)
return nil, err
}
if appInsertStmt == nil {
appInsertStmt, err = Flux.db.Prepare("INSERT INTO apps (name, deployment_id) VALUES ($1, $2) RETURNING id, name, deployment_id")
if err != nil {
return nil, fmt.Errorf("Failed to prepare statement: %v", err)
}
}
// create app in the database
err = appInsertStmt.QueryRow(projectConfig.Name, deployment.ID).Scan(&app.ID, &app.Name, &app.DeploymentID)
if err != nil {
return nil, fmt.Errorf("Failed to insert app: %v", err)
}
err = deployment.Start(ctx)
if err != nil {
return nil, fmt.Errorf("Failed to start deployment: %v", err)
}
var headContainer *Container
for _, container := range deployment.Containers {
if container.Head {
headContainer = &container
}
}
deployment.Proxy, err = NewDeploymentProxy(&deployment, headContainer)
if err != nil {
return nil, fmt.Errorf("Failed to create deployment proxy: %v", err)
}
Flux.proxy.AddDeployment(&deployment)
Flux.appManager.AddApp(app.Name, app)
return app, nil
}
func (app *App) Upgrade(ctx context.Context, projectConfig pkg.ProjectConfig, imageName string, projectPath string) error {
log.Printf("Upgrading deployment %s...\n", app.Name)
// if deploy is not started, start it
deploymentStatus, err := app.Deployment.Status(ctx)
if deploymentStatus != "running" || err != nil {
err = app.Deployment.Start(ctx)
if err != nil {
return fmt.Errorf("Failed to start deployment: %v", err)
}
}
err = app.Deployment.Upgrade(ctx, projectConfig, imageName, projectPath)
if err != nil {
return fmt.Errorf("Failed to upgrade deployment: %v", err)
}
return nil
}
func (app *App) Remove(ctx context.Context) error {
err := app.Deployment.Remove(ctx)
if err != nil {
log.Printf("Failed to remove deployment: %v\n", err)
return err
}
_, err = Flux.db.Exec("DELETE FROM apps WHERE id = ?", app.ID)
if err != nil {
log.Printf("Failed to delete app: %v\n", err)
return err
}
projectPath := filepath.Join(Flux.rootDir, "apps", app.Name)
err = os.RemoveAll(projectPath)
if err != nil {
return fmt.Errorf("Failed to remove project directory: %v", err)
}
return nil
}
type AppManager struct {
sync.Map
}
func (am *AppManager) GetApp(name string) *App {
app, exists := am.Load(name)
if !exists {
return nil
}
return app.(*App)
}
func (am *AppManager) GetAllApps() []*App {
var apps []*App
am.Range(func(key, value interface{}) bool {
if app, ok := value.(*App); ok {
apps = append(apps, app)
}
return true
})
return apps
}
func (am *AppManager) AddApp(name string, app *App) {
am.Store(name, app)
}
func (am *AppManager) DeleteApp(name string) error {
app := am.GetApp(name)
if app == nil {
return fmt.Errorf("App not found")
}
err := app.Remove(context.Background())
if err != nil {
return err
}
am.Delete(name)
return nil
}
func (am *AppManager) Init(db *sql.DB) {
log.Printf("Initializing deployments...\n")
if db == nil {
log.Panicf("DB is nil")
}
rows, err := db.Query("SELECT id, name, deployment_id FROM apps")
if err != nil {
log.Printf("Failed to query apps: %v\n", err)
return
}
defer rows.Close()
var apps []App
for rows.Next() {
var app App
if err := rows.Scan(&app.ID, &app.Name, &app.DeploymentID); err != nil {
log.Printf("Failed to scan app: %v\n", err)
return
}
apps = append(apps, app)
}
for _, app := range apps {
var deployment Deployment
var headContainer *Container
db.QueryRow("SELECT id, url, port FROM deployments WHERE id = ?", app.DeploymentID).Scan(&deployment.ID, &deployment.URL, &deployment.Port)
deployment.Containers = make([]Container, 0)
rows, err = db.Query("SELECT id, container_id, deployment_id, head FROM containers WHERE deployment_id = ?", app.DeploymentID)
if err != nil {
log.Printf("Failed to query containers: %v\n", err)
return
}
defer rows.Close()
for rows.Next() {
var container Container
var containerIDString string
rows.Scan(&container.ID, &containerIDString, &container.DeploymentID, &container.Head)
container.Deployment = &deployment
copy(container.ContainerID[:], containerIDString)
if container.Head {
headContainer = &container
}
deployment.Containers = append(deployment.Containers, container)
}
for i, container := range deployment.Containers {
var volumes []Volume
rows, err := db.Query("SELECT id, volume_id, container_id FROM volumes WHERE container_id = ?", container.ID)
if err != nil {
log.Printf("Failed to query volumes: %v\n", err)
return
}
defer rows.Close()
for rows.Next() {
var volume Volume
rows.Scan(&volume.ID, &volume.VolumeID, &volume.ContainerID)
volumes = append(volumes, volume)
}
deployment.Containers[i].Volumes = volumes
}
deployment.Proxy, _ = NewDeploymentProxy(&deployment, headContainer)
app.Deployment = deployment
am.AddApp(app.Name, &app)
}
}

View File

@@ -15,7 +15,7 @@ import (
"github.com/docker/docker/api/types/volume"
"github.com/docker/docker/client"
"github.com/joho/godotenv"
"github.com/juls0730/fluxd/pkg"
"github.com/juls0730/flux/pkg"
)
var dockerClient *client.Client
@@ -45,7 +45,7 @@ func init() {
}
}
func CreateVolume(ctx context.Context, name string) (vol *Volume, err error) {
func CreateDockerVolume(ctx context.Context, name string) (vol *Volume, err error) {
dockerVolume, err := dockerClient.VolumeCreate(ctx, volume.CreateOptions{
Driver: "local",
DriverOpts: map[string]string{},
@@ -86,7 +86,7 @@ func CreateDockerContainer(ctx context.Context, imageName, projectPath string, p
}
}
vol, err := CreateVolume(ctx, fmt.Sprintf("flux_%s-volume", projectConfig.Name))
vol, err := CreateDockerVolume(ctx, fmt.Sprintf("flux_%s-volume", projectConfig.Name))
log.Printf("Creating container %s...\n", containerName)
resp, err := dockerClient.ContainerCreate(ctx, &container.Config{
@@ -177,6 +177,15 @@ func (c *Container) Wait(ctx context.Context, port uint16) error {
return WaitForDockerContainer(ctx, string(c.ContainerID[:]), port)
}
func (c *Container) Status(ctx context.Context) (string, error) {
containerJSON, err := dockerClient.ContainerInspect(ctx, string(c.ContainerID[:]))
if err != nil {
return "", err
}
return containerJSON.State.Status, nil
}
// RemoveContainer stops and removes a container, but be warned that this will not remove the container from the database
func RemoveDockerContainer(ctx context.Context, containerID string) error {
if err := dockerClient.ContainerStop(ctx, containerID, container.StopOptions{}); err != nil {

View File

@@ -9,7 +9,7 @@ import (
"net/http"
"os/exec"
"github.com/juls0730/fluxd/pkg"
"github.com/juls0730/flux/pkg"
)
var (
@@ -98,84 +98,16 @@ func (s *FluxServer) DeployHandler(w http.ResponseWriter, r *http.Request) {
app := Flux.appManager.GetApp(projectConfig.Name)
if app == nil {
app = &App{
Name: projectConfig.Name,
}
log.Printf("Creating deployment %s...\n", app.Name)
container, err := CreateDockerContainer(r.Context(), imageName, projectPath, projectConfig)
if err != nil || container == nil {
log.Printf("Failed to create container: %v\n", err)
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
deployment, err := CreateDeployment(*container, projectConfig.Port, projectConfig.Url, s.db)
app.Deployment = deployment
app, err = CreateApp(r.Context(), imageName, projectPath, projectConfig)
if err != nil {
log.Printf("Failed to create deployment: %v\n", err)
log.Printf("Failed to create app: %v", err)
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
if appInsertStmt == nil {
appInsertStmt, err = s.db.Prepare("INSERT INTO apps (name, deployment_id) VALUES ($1, $2) RETURNING id, name, deployment_id")
if err != nil {
log.Printf("Failed to prepare statement: %v\n", err)
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
}
// create app in the database
err = appInsertStmt.QueryRow(projectConfig.Name, deployment.ID).Scan(&app.ID, &app.Name, &app.DeploymentID)
if err != nil {
log.Printf("Failed to insert app: %v\n", err)
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
err = deployment.Start(r.Context())
if err != nil {
log.Printf("Failed to start deployment: %v\n", err)
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
var headContainer *Container
for _, container := range deployment.Containers {
if container.Head {
headContainer = &container
}
}
deployment.Proxy, err = NewDeploymentProxy(&deployment, headContainer)
if err != nil {
log.Printf("Failed to create deployment proxy: %v\n", err)
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
Flux.proxy.AddDeployment(&deployment)
Flux.appManager.AddApp(app.Name, app)
} else {
log.Printf("Upgrading deployment %s...\n", app.Name)
// if deploy is not started, start it
deploymentStatus, err := app.Deployment.Status(r.Context())
if deploymentStatus != "running" || err != nil {
err = app.Deployment.Start(r.Context())
if err != nil {
log.Printf("Failed to start deployment: %v\n", err)
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
}
err = app.Deployment.Upgrade(r.Context(), projectConfig, imageName, projectPath)
err = app.Upgrade(r.Context(), projectConfig, imageName, projectPath)
if err != nil {
log.Printf("Failed to upgrade deployment: %v\n", err)
log.Printf("Failed to upgrade deployment: %v", err)
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
@@ -214,6 +146,17 @@ func (s *FluxServer) StartDeployHandler(w http.ResponseWriter, r *http.Request)
return
}
if app.Deployment.Proxy == nil {
var headContainer *Container
for _, container := range app.Deployment.Containers {
if container.Head {
headContainer = &container
}
}
app.Deployment.Proxy, _ = NewDeploymentProxy(&app.Deployment, headContainer)
}
w.WriteHeader(http.StatusOK)
}

View File

@@ -5,11 +5,8 @@ import (
"database/sql"
"fmt"
"log"
"os"
"path/filepath"
"sync"
"github.com/juls0730/fluxd/pkg"
"github.com/juls0730/flux/pkg"
)
var (
@@ -19,39 +16,6 @@ var (
updateVolumeStmt *sql.Stmt
)
type AppManager struct {
sync.Map
}
type App struct {
ID int64 `json:"id,omitempty"`
Deployment Deployment `json:"-"`
Name string `json:"name,omitempty"`
DeploymentID int64 `json:"deployment_id,omitempty"`
}
func (app *App) Remove(ctx context.Context) error {
err := app.Deployment.Remove(ctx)
if err != nil {
log.Printf("Failed to remove deployment: %v\n", err)
return err
}
_, err = Flux.db.Exec("DELETE FROM apps WHERE id = ?", app.ID)
if err != nil {
log.Printf("Failed to delete app: %v\n", err)
return err
}
projectPath := filepath.Join(Flux.rootDir, "apps", app.Name)
err = os.RemoveAll(projectPath)
if err != nil {
return fmt.Errorf("Failed to remove project directory: %v", err)
}
return nil
}
type Deployment struct {
ID int64 `json:"id"`
Containers []Container `json:"-"`
@@ -60,127 +24,6 @@ type Deployment struct {
Port uint16 `json:"port"`
}
func (am *AppManager) GetApp(name string) *App {
app, exists := am.Load(name)
if !exists {
return nil
}
return app.(*App)
}
func (am *AppManager) GetAllApps() []*App {
var apps []*App
am.Range(func(key, value interface{}) bool {
if app, ok := value.(*App); ok {
apps = append(apps, app)
}
return true
})
return apps
}
func (am *AppManager) AddApp(name string, app *App) {
am.Store(name, app)
}
func (am *AppManager) DeleteApp(name string) error {
app := am.GetApp(name)
if app == nil {
return fmt.Errorf("App not found")
}
err := app.Remove(context.Background())
if err != nil {
return err
}
am.Delete(name)
return nil
}
func (am *AppManager) Init(db *sql.DB) {
log.Printf("Initializing deployments...\n")
if db == nil {
log.Panicf("DB is nil")
}
rows, err := db.Query("SELECT id, name, deployment_id FROM apps")
if err != nil {
log.Printf("Failed to query apps: %v\n", err)
return
}
defer rows.Close()
var apps []App
for rows.Next() {
var app App
if err := rows.Scan(&app.ID, &app.Name, &app.DeploymentID); err != nil {
log.Printf("Failed to scan app: %v\n", err)
return
}
apps = append(apps, app)
}
for _, app := range apps {
var deployment Deployment
var headContainer *Container
db.QueryRow("SELECT id, url, port FROM deployments WHERE id = ?", app.DeploymentID).Scan(&deployment.ID, &deployment.URL, &deployment.Port)
deployment.Containers = make([]Container, 0)
rows, err = db.Query("SELECT id, container_id, deployment_id, head FROM containers WHERE deployment_id = ?", app.DeploymentID)
if err != nil {
log.Printf("Failed to query containers: %v\n", err)
return
}
defer rows.Close()
for rows.Next() {
var container Container
var containerIDString string
rows.Scan(&container.ID, &containerIDString, &container.DeploymentID, &container.Head)
container.Deployment = &deployment
copy(container.ContainerID[:], containerIDString)
if container.Head {
headContainer = &container
}
deployment.Containers = append(deployment.Containers, container)
}
for i, container := range deployment.Containers {
var volumes []Volume
rows, err := db.Query("SELECT id, volume_id, container_id FROM volumes WHERE container_id = ?", container.ID)
if err != nil {
log.Printf("Failed to query volumes: %v\n", err)
return
}
defer rows.Close()
for rows.Next() {
var volume Volume
rows.Scan(&volume.ID, &volume.VolumeID, &volume.ContainerID)
volumes = append(volumes, volume)
}
deployment.Containers[i].Volumes = volumes
}
deployment.Proxy, err = NewDeploymentProxy(&deployment, headContainer)
if err != nil {
log.Printf("Failed to create deployment proxy: %v\n", err)
return
}
app.Deployment = deployment
am.AddApp(app.Name, &app)
}
}
// Creates a deployment and containers in the database
func CreateDeployment(container Container, port uint16, appUrl string, db *sql.DB) (Deployment, error) {
var deployment Deployment
@@ -301,7 +144,7 @@ func (deployment *Deployment) Upgrade(ctx context.Context, projectConfig pkg.Pro
copy(container.ContainerID[:], containerIDString)
deployment.Containers = append(deployment.Containers, container)
log.Printf("Starting container %s...\n", container.ContainerID[:])
log.Printf("Starting container %s...\n", container.ContainerID[:12])
err = container.Start(ctx)
if err != nil {
log.Printf("Failed to start container: %v\n", err)
@@ -313,26 +156,8 @@ func (deployment *Deployment) Upgrade(ctx context.Context, projectConfig pkg.Pro
return err
}
tx, err := Flux.db.Begin()
if err != nil {
log.Printf("Failed to begin transaction: %v\n", err)
return err
}
if _, err := tx.Exec("UPDATE deployments SET url = ?, port = ? WHERE id = ?", projectConfig.Url, projectConfig.Port, deployment.ID); err != nil {
if _, err := Flux.db.Exec("UPDATE deployments SET url = ?, port = ? WHERE id = ?", projectConfig.Url, projectConfig.Port, deployment.ID); err != nil {
log.Printf("Failed to update deployment: %v\n", err)
tx.Rollback()
return err
}
if _, err := tx.Exec("UPDATE apps SET deployment_id = ? WHERE name = ?", deployment.ID, projectConfig.Name); err != nil {
log.Printf("Failed to update app: %v\n", err)
tx.Rollback()
return err
}
if err := tx.Commit(); err != nil {
log.Printf("Failed to commit transaction: %v\n", err)
return err
}
@@ -344,7 +169,7 @@ func (deployment *Deployment) Upgrade(ctx context.Context, projectConfig pkg.Pro
return err
}
tx, err = Flux.db.Begin()
tx, err := Flux.db.Begin()
if err != nil {
log.Printf("Failed to begin transaction: %v\n", err)
return err
@@ -354,7 +179,7 @@ func (deployment *Deployment) Upgrade(ctx context.Context, projectConfig pkg.Pro
var oldContainers []*Container
for _, container := range deployment.Containers {
if existingContainers[string(container.ContainerID[:])] {
log.Printf("Deleting container from db: %s\n", container.ContainerID[0:12])
log.Printf("Deleting container from db: %s\n", container.ContainerID[:12])
_, err = tx.Exec("DELETE FROM containers WHERE id = ?", container.ID)
oldContainers = append(oldContainers, &container)
@@ -393,16 +218,6 @@ func (deployment *Deployment) Upgrade(ctx context.Context, projectConfig pkg.Pro
return nil
}
func arrayContains(arr []string, str string) bool {
for _, a := range arr {
if a == str {
return true
}
}
return false
}
func (d *Deployment) Remove(ctx context.Context) error {
for _, container := range d.Containers {
err := container.Remove(ctx)
@@ -445,15 +260,6 @@ func (d *Deployment) Stop(ctx context.Context) error {
return nil
}
func (c *Container) GetStatus(ctx context.Context) (string, error) {
containerJSON, err := dockerClient.ContainerInspect(ctx, string(c.ContainerID[:]))
if err != nil {
return "", err
}
return containerJSON.State.Status, nil
}
func (d *Deployment) Status(ctx context.Context) (string, error) {
var status string
if d == nil {
@@ -467,7 +273,7 @@ func (d *Deployment) Status(ctx context.Context) (string, error) {
}
for _, container := range d.Containers {
containerStatus, err := container.GetStatus(ctx)
containerStatus, err := container.Status(ctx)
if err != nil {
log.Printf("Failed to get container status: %v\n", err)
return "", err

View File

@@ -50,7 +50,7 @@ func NewDeploymentProxy(deployment *Deployment, head *Container) (*DeploymentPro
}
if containerJSON.NetworkSettings.IPAddress == "" {
return nil, fmt.Errorf("No IP address found for container %s", head.ContainerID[0:12])
return nil, fmt.Errorf("No IP address found for container %s", head.ContainerID[:12])
}
containerUrl, err := url.Parse(fmt.Sprintf("http://%s:%d", containerJSON.NetworkSettings.IPAddress, deployment.Port))

View File

@@ -14,7 +14,7 @@ import (
_ "embed"
"github.com/juls0730/fluxd/pkg"
"github.com/juls0730/flux/pkg"
_ "github.com/mattn/go-sqlite3"
)