houston/server/server.go

package server

import (
	"encoding/json"
	"fmt"
	"net"
	"os"
	"sync/atomic"

	"repositories.action2quare.com/ayo/gocommon/logger"
	"repositories.action2quare.com/ayo/houston/client"
	"repositories.action2quare.com/ayo/houston/shared"
	"repositories.action2quare.com/ayo/houston/shared/protos"

	"google.golang.org/grpc"
)

// protoc --go_out=. --go-grpc_out=. protos/*.proto
type HoustonServer interface {
	Start() error
	Stop()
	Operation() Operation
}

type serverConfig struct {
	GrpcPort    int    `json:"grpc_port"`
	StorageRoot string `json:"storage_path"`
	RunAsClient bool   `json:"run_as_client"`
}

type DeployRequest struct {
	shared.DeployRequest
	hostnames []string
}

func MakeDeployRequest(req shared.DeployRequest, targets []string) DeployRequest {
	return DeployRequest{
		DeployRequest: req,
		hostnames:     targets,
	}
}

type WithdrawRequest struct {
	shared.WithdrawRequest
	hostnames []string
}

func MakeWithdrawRequest(req shared.WithdrawRequest, targets []string) WithdrawRequest {
	return WithdrawRequest{
		WithdrawRequest: req,
		hostnames:       targets,
	}
}

type StartProcessRequest struct {
	shared.StartProcessRequest
	hostnames []string
}

func MakeStartProcessRequest(req shared.StartProcessRequest, targets []string) StartProcessRequest {
	return StartProcessRequest{
		StartProcessRequest: req,
		hostnames:           targets,
	}
}

type StopProcessRequest struct {
	shared.StopProcessRequest
	hostnames []string
}

func MakeStopProcessRequest(req shared.StopProcessRequest, targets []string) StopProcessRequest {
	return StopProcessRequest{
		StopProcessRequest: req,
		hostnames:          targets,
	}
}

type RestartProcessRequest struct {
	shared.RestartProcessRequest
	hostnames []string
}

func MakeRestartRequest(req shared.RestartProcessRequest, targets []string) RestartProcessRequest {
	return RestartProcessRequest{
		RestartProcessRequest: req,
		hostnames:             targets,
	}
}

type UploadRequest struct {
	shared.UploadRequest
	hostnames []string
}

func MakeUploadRequest(req shared.UploadRequest, targets []string) UploadRequest {
	return UploadRequest{
		UploadRequest: req,
		hostnames:     targets,
	}
}

type Operation interface {
	Deploy(DeployRequest)
	Withdraw(WithdrawRequest)
	StartProcess(StartProcessRequest)
	StopProcess(StopProcessRequest)
	RestartProcess(RestartProcessRequest)
	Upload(UploadRequest)
	Hosts() map[string]hostSnapshot
}

func loadServerConfig() serverConfig {
	configFile, err := os.Open("config.json")
	if err != nil {
		logger.Println(err)
		return serverConfig{
			GrpcPort: 8080,
		}
	}
	defer configFile.Close()

	var config struct {
		Houston *struct {
			Server serverConfig `json:"server"`
		} `json:"houston"`
	}

	dec := json.NewDecoder(configFile)
	err = dec.Decode(&config)
	if err != nil {
		logger.Println(err)
		return serverConfig{
			GrpcPort: 8080,
		}
	}

	if config.Houston == nil {
		logger.Println(`"houston" object is missing in config.json`)
		return serverConfig{
			GrpcPort: 8080,
		}
	}

	return config.Houston.Server
}

func NewServer() HoustonServer {
	var opts []grpc.ServerOption
	grpcServer := grpc.NewServer(opts...)

	os := newOperationServer()
	ms := newMonitorServer()
	protos.RegisterOperationServer(grpcServer, os)
	protos.RegisterMonitorServer(grpcServer, ms)

	return &houstonServer{
		rpcServer: grpcServer,
		os:        os,
		ms:        ms,
		port:      loadServerConfig().GrpcPort,
	}
}

type houstonServer struct {
	rpcServer *grpc.Server
	os        *operationServer
	ms        *monitorServer
	port      int
}

func (hs *houstonServer) Start() error {
	logger.Println("houston server is started at port", hs.port)
	lis, err := net.Listen("tcp", fmt.Sprintf("0.0.0.0:%d", hs.port))
	if err != nil {
		return err
	}

	closeCount := int32(0)
	var hc client.HoustonClient
	if loadServerConfig().RunAsClient {
		hc, err = client.NewClient(false)
		if err != nil {
			return err
		}

		go func() {
			hc.Start()
			logger.Println("houstonClient is finished")
			if atomic.AddInt32(&closeCount, 1) == 1 {
				logger.Println("try stop houstonServer")
				hs.Stop()
			}
		}()
	}

	err = hs.rpcServer.Serve(lis)
	if atomic.AddInt32(&closeCount, 1) == 1 {
		if hc != nil {
			logger.Println("try stop houstonClient")
			hc.Shutdown()
		}
	}
	logger.Println("houstonServer is finished")

	return err
}

func (hs *houstonServer) Stop() {
	hs.rpcServer.GracefulStop()
}

func (hs *houstonServer) Operation() Operation {
	return hs.os
}