core/oc-monitord
5
0
Fork 0
Code Issues 2 Pull Requests Packages Projects Releases Wiki Activity

Minimize code + docker + monitord naming

Browse Source
This commit is contained in:
mr 2024-08-19 11:43:40 +02:00
parent 99bdc1b784
commit feca395039
22 changed files with 150 additions and 751 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
Dockerfile
View File

@ -4,15 +4,11 @@ WORKDIR /app
COPY . .
RUN apt install tar/stable
RUN go build .
RUN mkdir workflows/
FROM scratch
FROM argoproj/argocd:latest
WORKDIR /app
COPY --from=builder /bin/tar /bin/tar
COPY conf/ocmonitor_conf.json /app/conf/
COPY --from=builder /app/oc-monitor .
COPY --from=builder /app/workflows/ .
ENTRYPOINT [ "/app/oc-monitor" ]
COPY conf/docker_ocmonitord_conf.json /app/conf/
COPY --from=builder /app/oc-monitord .

6
README.md
View File

@ -4,12 +4,12 @@
While a registry with all of the OC docker images has not been set-up we can export this image to k3s ctr
> docker save oc-monitor:latest | sudo k3s ctr images import -
> docker save oc-monitord:latest | sudo k3s ctr images import -
Then in the pod manifest for oc-monitor use :
Then in the pod manifest for oc-monitord use :
```
image: docker.io/library/oc-monitor
image: docker.io/library/oc-monitord
imagePullPolicy: Never
```

46
argo_workflows/chalitic-scobeele_01_08_2024_183304.yml
View File

@ -1,46 +0,0 @@
apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
generateName: oc-test-chalitic-scobeele
spec:
entrypoint: dag
volumeClaimTemplates:
- metadata:
name: workdir
spec:
accessModes: [ReadWriteOnce]
resources:
requests:
storage: 1Gi
templates:
- name: "-0"
container:
image: ""
args: [""]
volumeMounts:
- name: workdir
mountPath: /mnt/vol
- name: curl-1
container:
image: curlimages/curl:7.88.1
args: [-SL, 'https://toulousefc.com', -o, /mnt/vol/tfc.hmtl]
volumeMounts:
- name: workdir
mountPath: /mnt/vol
- name: alpine-2
container:
image: alpine:3.7
command: [sh, -c]
args: ['grep ''<meta name="description"'' /mnt/vol/tfc.html | awk -F ''content="'' ''{print $2}'' | awk -F ''"'' ''{print $1}''']
volumeMounts:
- name: workdir
mountPath: /mnt/vol
- name: dag
dag:
tasks:
- name: "-0"
template: "-0"
- name: curl-1
template: curl-1
- name: alpine-2
template: alpine-2

46
argo_workflows/colintic-schicke_01_08_2024_183508.yml
View File

@ -1,46 +0,0 @@
apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
generateName: oc-test-colintic-schicke
spec:
entrypoint: dag
volumeClaimTemplates:
- metadata:
name: workdir
spec:
accessModes: [ReadWriteOnce]
resources:
requests:
storage: 1Gi
templates:
- name: "-0"
container:
image: ""
args: [""]
volumeMounts:
- name: workdir
mountPath: /mnt/vol
- name: curl-1
container:
image: curlimages/curl:7.88.1
args: [-SL, 'https://toulousefc.com', -o, /mnt/vol/tfc.hmtl]
volumeMounts:
- name: workdir
mountPath: /mnt/vol
- name: alpine-2
container:
image: alpine:3.7
command: [sh, -c]
args: ['grep ''<meta name="description"'' /mnt/vol/tfc.html | awk -F ''content="'' ''{print $2}'' | awk -F ''"'' ''{print $1}''']
volumeMounts:
- name: workdir
mountPath: /mnt/vol
- name: dag
dag:
tasks:
- name: "-0"
template: "-0"
- name: curl-1
template: curl-1
- name: alpine-2
template: alpine-2

39
argo_workflows/sismon-oximous_02_08_2024_131457.yml
View File

@ -1,39 +0,0 @@
apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
generateName: oc-test-sismon-oximous
spec:
entrypoint: dag
volumeClaimTemplates:
- metadata:
name: workdir
spec:
accessModes: [ReadWriteOnce]
resources:
requests:
storage: 1Gi
templates:
- name: alpine-b6f790ae-5db2-4458-8375-59450f12f65b
container:
image: alpine:3.7
command: [sh, -c]
args: ['grep ''<meta name="description"'' /mnt/vol/tfc.html | awk -F ''content="'' ''{print $2}'' | awk -F ''"'' ''{print $1}''']
volumeMounts:
- name: workdir
mountPath: /mnt/vol
- name: curl-aa8d2265-9fe2-42c7-ba1f-46ea0da8e633
container:
image: curlimages/curl:7.88.1
args: [-SL, 'https://toulousefc.com', -o, /mnt/vol/tfc.hmtl]
volumeMounts:
- name: workdir
mountPath: /mnt/vol
- name: dag
dag:
tasks:
- name: curl-aa8d2265-9fe2-42c7-ba1f-46ea0da8e633
template: curl-aa8d2265-9fe2-42c7-ba1f-46ea0da8e633
- name: alpine-b6f790ae-5db2-4458-8375-59450f12f65b
template: alpine-b6f790ae-5db2-4458-8375-59450f12f65b
dependencies:
- curl-aa8d2265-9fe2-42c7-ba1f-46ea0da8e633

46
argo_workflows/spresis-aplaront_01_08_2024_182306.yml
View File

@ -1,46 +0,0 @@
apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
generateName: oc-test-spresis-aplaront
spec:
entrypoint: dag
volumeClaimTemplates:
- metadata:
name: workdir
spec:
accessModes: [ReadWriteOnce]
resources:
requests:
storage: 1Gi
templates:
- name: "-0"
container:
image: ""
args: [""]
volumeMounts:
- name: workdir
mountPath: /mnt/vol
- name: curl-1
container:
image: curlimages/curl:7.88.1
args: [-SL, 'https://toulousefc.com', -o, /mnt/vol/tfc.hmtl]
volumeMounts:
- name: workdir
mountPath: /mnt/vol
- name: alpine-2
container:
image: alpine:3.7
command: [sh, -c]
args: ['grep ''<meta name="description"'' /mnt/vol/tfc.html | awk -F ''content="'' ''{print $2}'' | awk -F ''"'' ''{print $1}''']
volumeMounts:
- name: workdir
mountPath: /mnt/vol
- name: dag
dag:
tasks:
- name: '-'
template: '-'
- name: curl-0d565c87-50ae-4a73-843d-f8b2d4047772
template: curl-0d565c87-50ae-4a73-843d-f8b2d4047772
- name: alpine-2ce0323f-a85d-4b8b-a783-5280f48d634a
template: alpine-2ce0323f-a85d-4b8b-a783-5280f48d634a

10
conf/conf.go
View File

@ -3,13 +3,9 @@ package conf
import "sync"
type Config struct {
LokiURL string
ExecutionID string
WorkflowID string
OcCatalogUrl string
MongoUrl string
Database string
MonitorDir string
LokiURL string
ExecutionID string
WorkflowID string
}
var instance *Config

0
conf/docker_ocmonitor_conf.json → conf/docker_ocmonitord_conf.json
View File

0
conf/local_ocmonitor_conf.json → conf/local_ocmonitord_conf.json
View File

2
go.mod
View File

@ -1,4 +1,4 @@
module oc-monitor
module oc-monitord
go 1.22.0

15
logger/logger.go
View File

@ -1,15 +0,0 @@
package logger
import (
"os"
"time"
"github.com/rs/zerolog"
)
var Logger zerolog.Logger
func init() {
output := zerolog.ConsoleWriter{Out: os.Stdout, TimeFormat: time.RFC3339}
Logger = zerolog.New(output).With().Timestamp().Logger()
}

78
loki_logger.go
View File

@ -1,78 +0,0 @@
package main
import (
"bytes"
"encoding/json"
"fmt"
"net/http"
"time"
)
type LokiLogger struct {
url string
client *http.Client
}
func NewLokiLogger(url string) *LokiLogger {
return &LokiLogger{
url: url,
client: &http.Client{Timeout: 10 * time.Second},
}
}
type logEntry struct {
Timestamp int64 `json:"ts"`
Line string `json:"line"`
}
type logStream struct {
Labels string `json:"labels"`
Entries []logEntry `json:"entries"`
}
type pushRequest struct {
Streams []logStream `json:"streams"`
}
func (l *LokiLogger) Log(label string, message string) error {
entry := logEntry{
Timestamp: time.Now().UnixNano(),
Line: message,
}
stream := logStream{
Labels: label,
Entries: []logEntry{entry},
}
reqBody := pushRequest{
Streams: []logStream{stream},
}
body, err := json.Marshal(reqBody)
if err != nil {
return err
}
req, err := http.NewRequest("POST", l.url, bytes.NewBuffer(body))
if err != nil {
return err
}
req.Header.Set("Content-Type", "application/json")
resp, err := l.client.Do(req)
if err != nil {
return err
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusNoContent {
return fmt.Errorf("unexpected status code: %d", resp.StatusCode)
}
return nil
}
func (l *LokiLogger) Close() {
// No cleanup needed for pure HTTP client
}

147
main.go
View File

@ -11,9 +11,9 @@ import (
"strings"
"sync"
"oc-monitor/conf"
"oc-monitor/models"
"oc-monitor/workflow_builder"
"oc-monitord/conf"
"oc-monitord/models"
"oc-monitord/workflow_builder"
oclib "cloud.o-forge.io/core/oc-lib"
@ -26,53 +26,47 @@ import (
"github.com/rs/zerolog"
)
var logger zerolog.Logger
var wf_logger zerolog.Logger
var pods_logger zerolog.Logger
var parser argparse.Parser
var monitorLocal bool
var workflowName string
var logger zerolog.Logger
var wf_logger zerolog.Logger
var pods_logger zerolog.Logger
var parser argparse.Parser
var monitorLocal bool
var workflowName string
const defaultConfigFile = "/etc/oc/ocmonitor_conf.json"
const localConfigFile = "./conf/ocmonitor_conf.json"
const defaultConfigFile = "/etc/oc/ocmonitord_conf.json"
const localConfigFile = "./conf/local_ocmonitord_conf.json"
func main() {
monitorLocal = false
// Test if monitor is launched outside (with parameters) or in a k8s environment (env variables sets)
if os.Getenv("KUBERNETES_SERVICE_HOST") == ""{
if os.Getenv("KUBERNETES_SERVICE_HOST") == "" {
// Not in a k8s environment, get conf from parameters
fmt.Println("Executes outside of k8s")
parser = *argparse.NewParser("oc-monitor","Launch the execution of a workflow given as a parameter and sends the produced logs to a loki database")
parser = *argparse.NewParser("oc-monitord", "Launch the execution of a workflow given as a parameter and sends the produced logs to a loki database")
loadConfig(false, &parser)
} else {
// Executed in a k8s environment
} else {
// Executed in a k8s environment
fmt.Println("Executes inside a k8s")
monitorLocal = true
loadConfig(true,nil)
loadConfig(true, nil)
}
logs.SetAppName("oc-monitor")
logger = logs.CreateLogger("oc-monitor", conf.GetConfig().LokiURL)
logs.SetAppName("oc-monitord")
logger = logs.CreateLogger("oc-monitord", conf.GetConfig().LokiURL)
logger.Debug().Msg("Loki URL : " + conf.GetConfig().LokiURL)
logger.Debug().Msg("Workflow executed : " + conf.GetConfig().ExecutionID)
oclib.SetConfig(conf.GetConfig().MongoUrl,conf.GetConfig().Database)
oclib.Init("oc-monitor")
oclib.Init("oc-monitord")
wf_id := getWorkflowId(conf.GetConfig().ExecutionID)
conf.GetConfig().WorkflowID = wf_id
logger.Debug().Msg("Starting construction of yaml argo for workflow :" + wf_id)
if _, err := os.Stat(conf.GetConfig().MonitorDir + "/argo_workflows/"); os.IsNotExist(err) {
os.Mkdir(conf.GetConfig().MonitorDir + "/argo_workflows/",0755)
if _, err := os.Stat("./argo_workflows/"); os.IsNotExist(err) {
os.Mkdir("./argo_workflows/", 0755)
logger.Info().Msg("Created argo_workflows/")
}
@ -86,14 +80,14 @@ func main() {
argo_file_path, err := new_wf.ExportToArgo()
if err != nil {
logger.Error().Msg("Could not create the Argo file for " + conf.GetConfig().WorkflowID )
logger.Error().Msg("Could not create the Argo file for " + conf.GetConfig().WorkflowID)
logger.Error().Msg(err.Error())
}
logger.Debug().Msg("Created :" + argo_file_path)
workflowName = getContainerName(argo_file_path)
wf_logger = logger.With().Str("argo_name", workflowName).Str("workflow_id",conf.GetConfig().WorkflowID).Str("workflow_execution_id",conf.GetConfig().ExecutionID).Logger()
wf_logger = logger.With().Str("argo_name", workflowName).Str("workflow_id", conf.GetConfig().WorkflowID).Str("workflow_execution_id", conf.GetConfig().ExecutionID).Logger()
wf_logger.Debug().Msg("Testing argo name")
executeWorkflow(argo_file_path)
@ -103,7 +97,7 @@ func main() {
// Return the Workflow ID associated to a workflow execution object
func getWorkflowId(exec_id string) string {
res := oclib.LoadOne(oclib.LibDataEnum(oclib.WORKFLOW_EXECUTION),exec_id)
res := oclib.LoadOne(oclib.LibDataEnum(oclib.WORKFLOW_EXECUTION), exec_id)
if res.Code != 200 {
logger.Error().Msg("Could not retrieve workflow ID from execution ID " + exec_id)
@ -120,48 +114,23 @@ func executeWorkflow(argo_file_path string) {
var stdout, stderr io.ReadCloser
// var stderr io.ReadCloser
var err error
cmd := exec.Command("argo", "submit", "--watch", conf.GetConfig().MonitorDir + "/argo_workflows/"+argo_file_path, "--serviceaccount=argo", "-n","argo")
if stdout, err = cmd.StdoutPipe(); err != nil{
cmd := exec.Command("argo", "submit", "--watch", "./argo_workflows/"+argo_file_path, "--serviceaccount=argo", "-n", "argo")
if stdout, err = cmd.StdoutPipe(); err != nil {
wf_logger.Error().Msg("Could not retrieve stdoutpipe " + err.Error())
return
}
if stderr, err = cmd.StderrPipe(); err != nil{
if stderr, err = cmd.StderrPipe(); err != nil {
wf_logger.Error().Msg("Could not retrieve stderrpipe " + err.Error())
return
}
// cmd_logs := exec.Command("argo","logs","oc-monitor-"+workflowName,"-f" , "-n", "argo")
// // if stdout_logs, err = cmd_logs.StdoutPipe(); err != nil{
// // wf_logger.Error().Msg("Could not retrieve stdoutpipe " + err.Error())
// // return
// // }
// // if stderr_logs, err = cmd_logs.StderrPipe(); err != nil{
// // wf_logger.Error().Msg("Could not retrieve stderrpipe " + err.Error())
// // return
// // }
if err := cmd.Start(); err != nil {
panic(err)
}
// if err := cmd_logs.Start(); err != nil {
// panic(err)
// }
var wg sync.WaitGroup
go logWorkflow(stdout, &wg)
// go logWorkflow(stderr,&wg)
// time.Sleep(time.Second * 1)
// go logPods(stdout_logs,"out")
// go logPods(stderr_logs,"err")
if err := cmd.Wait(); err != nil {
wf_logger.Error().Msg("Could not execute argo submit")
@ -169,13 +138,6 @@ func executeWorkflow(argo_file_path string) {
updateStatus("fatal")
}
// if err := cmd_logs.Wait(); err != nil {
// wf_logger.Error().Msg("Could not execute argo logs")
// wf_logger.Error().Msg(err.Error() + bufio.NewScanner(stdout_logs).Text())
// wf_logger.Error().Msg(err.Error() + bufio.NewScanner(stderr_logs).Text())
// }
wg.Wait()
}
@ -184,17 +146,17 @@ func executeWorkflow(argo_file_path string) {
func logWorkflow(pipe io.ReadCloser, wg *sync.WaitGroup) {
var current_watch, previous_watch models.ArgoWatch
watch_output := make([]string,0)
watch_output := make([]string, 0)
scanner := bufio.NewScanner(pipe)
for scanner.Scan() {
log := scanner.Text()
watch_output = append(watch_output, log)
if(strings.HasPrefix(log, "Progress:")){
if strings.HasPrefix(log, "Progress:") {
current_watch = *models.NewArgoLogs(watch_output)
workflowName = current_watch.Name
if(!current_watch.Equals(previous_watch)){
if !current_watch.Equals(previous_watch) {
wg.Add(1)
checkStatus(current_watch.Status, previous_watch.Status)
jsonified, err := json.Marshal(current_watch)
@ -211,8 +173,8 @@ func logWorkflow(pipe io.ReadCloser, wg *sync.WaitGroup) {
}
// Debug, no logs sent
func logPods(pipe io.ReadCloser,name string){
pods_logger = wf_logger.With().Str("sortie name",name).Logger()
func logPods(pipe io.ReadCloser, name string) {
pods_logger = wf_logger.With().Str("sortie name", name).Logger()
scanner := bufio.NewScanner(pipe)
for scanner.Scan() {
log := scanner.Text()
@ -221,61 +183,50 @@ func logPods(pipe io.ReadCloser,name string){
}
func loadConfig(is_k8s bool, parser *argparse.Parser){
func loadConfig(is_k8s bool, parser *argparse.Parser) {
var o *onion.Onion
o = initOnion(o)
current_path, err := os.Getwd()
if err != nil {
logger.Error().Msg("Could not get the working directory path")
}
conf.GetConfig().OcCatalogUrl = o.GetStringDefault("oc-catalog","https://localhost:8087")
conf.GetConfig().MonitorDir = current_path
// These variables can only be retrieved in the onion
// Variables that don't depend on the environmen (from conf file), can be loaded after
// We can't use underscore in the env variable names because it's the delimitor with OCMONITOR too
setConf(is_k8s, o, parser)
if (!IsValidUUID(conf.GetConfig().ExecutionID)){
if !IsValidUUID(conf.GetConfig().ExecutionID) {
logger.Fatal().Msg("Provided ID is not an UUID")
}
}
func setConf(is_k8s bool, o *onion.Onion, parser *argparse.Parser) {
mongo := ""
db := ""
if is_k8s {
conf.GetConfig().LokiURL = o.GetStringDefault("lokiurl", "http://127.0.0.1:3100")
conf.GetConfig().ExecutionID = o.GetString("workflow")
conf.GetConfig().MongoUrl = o.GetStringDefault("mongourl", "mongodb://127.0.0.1:27017")
conf.GetConfig().Database = o.GetStringDefault("database", "DC_myDC")
mongo = o.GetStringDefault("mongourl", "mongodb://127.0.0.1:27017")
db = o.GetStringDefault("database", "DC_myDC")
} else {
url := parser.String("u", "url", &argparse.Options{Required: true, Default: "http://127.0.0.1:3100", Help: "Url to the Loki database logs will be sent to"})
workflow := parser.String("w", "workflow", &argparse.Options{Required: true, Help: "Execution ID of the workflow to request from oc-catalog API"})
mongo := parser.String("m", "mongo", &argparse.Options{Required: true, Help: "URL to reach the MongoDB"})
db := parser.String("d", "database", &argparse.Options{Required: true, Help: "Name of the database to query in MongoDB"})
mongo = *parser.String("m", "mongo", &argparse.Options{Required: true, Help: "URL to reach the MongoDB"})
db = *parser.String("d", "database", &argparse.Options{Required: true, Help: "Name of the database to query in MongoDB"})
err := parser.Parse(os.Args)
if err != nil {
fmt.Println(parser.Usage(err))
os.Exit(1)
}
conf.GetConfig().LokiURL = *url
conf.GetConfig().ExecutionID = *workflow
conf.GetConfig().MongoUrl = *mongo
conf.GetConfig().Database = *db
}
oclib.SetConfig(mongo, db)
}
func initOnion(o *onion.Onion) *onion.Onion {
logger = logs.CreateLogger("oc-monitor", "")
logger = logs.CreateLogger("oc-monitord", "")
configFile := ""
l3 := onion.NewEnvLayerPrefix("_", "OCMONITOR")
l3 := onion.NewEnvLayerPrefix("_", "OCMONITORD")
l2, err := onion.NewFileLayer(defaultConfigFile, nil)
if err == nil {
logger.Info().Msg("Config file found : " + defaultConfigFile)
@ -300,11 +251,11 @@ func initOnion(o *onion.Onion) *onion.Onion {
}
func IsValidUUID(u string) bool {
_, err := uuid.Parse(u)
return err == nil
}
_, err := uuid.Parse(u)
return err == nil
}
func getContainerName(argo_file string) string {
func getContainerName(argo_file string) string {
regex := "([a-zA-Z]+-[a-zA-Z]+)"
re := regexp.MustCompile(regex)
@ -326,11 +277,11 @@ func updateStatus(status string) {
wf_exec.ArgoStatusToState(status)
serialized := wf_exec.Serialize()
res := oclib.UpdateOne(oclib.LibDataEnum(oclib.WORKFLOW_EXECUTION),serialized, exec_id)
res := oclib.UpdateOne(oclib.LibDataEnum(oclib.WORKFLOW_EXECUTION), serialized, exec_id)
if res.Code != 200 {
logger.Error().Msg("Could not update status for workflow execution " + exec_id)
}
fmt.Printf("status argo : %s /nstatus db : %s",status,serialized["state"])
fmt.Printf("status argo : %s /nstatus db : %s", status, serialized["state"])
}

21
models/argo_logs.go
View File

@ -5,19 +5,19 @@ import (
)
type ArgoWatch struct {
Name string
Namespace string
Status string
Name string
Namespace string
Status string
Conditions
Created string
Started string
Duration string
Progress string
Created string
Started string
Duration string
Progress string
}
type Conditions struct{
PodRunning bool
Completed bool
type Conditions struct {
PodRunning bool
Completed bool
}
func (a *ArgoWatch) Equals(arg ArgoWatch) bool {
@ -33,7 +33,6 @@ func NewArgoLogs(inputs []string) *ArgoWatch {
if line == "" {
continue
}
switch {
case strings.HasPrefix(line, "Name:"):
workflow.Name = parseValue(line)

53
models/http.go
View File

@ -1,53 +0,0 @@
package models
import (
"io"
"net/http"
"net/http/cookiejar"
"net/url"
)
type HttpQuery struct {
baseurl string
jar http.CookieJar
Cookies map[string]string
}
func (h *HttpQuery) Init(url string) {
h.baseurl = url
h.jar, _ = cookiejar.New(nil)
h.Cookies = make(map[string]string)
}
func (h *HttpQuery) Get(url string) ([]byte, error) {
client := &http.Client{Jar: h.jar}
resp, err := client.Get(h.baseurl + url)
if err != nil {
return nil, err
}
// store received cookies
for _, cookie := range h.jar.Cookies(resp.Request.URL) {
h.Cookies[cookie.Name] = cookie.Value
}
if err != nil {
return nil, err
}
body, err := io.ReadAll(resp.Body)
if err != nil {
return nil, err
}
return body, nil
}
func (h *HttpQuery) Post(url string, data url.Values) (*http.Response, error) {
client := &http.Client{Jar: h.jar}
resp, err := client.PostForm(h.baseurl+url, data)
if err != nil {
return nil, err
}
// store received cookies
for _, cookie := range h.jar.Cookies(resp.Request.URL) {
h.Cookies[cookie.Name] = cookie.Value
}
return resp, err
}

74
models/schedule.go
View File

@ -1,74 +0,0 @@
package models
import (
"fmt"
"oc-monitor/logger"
"sync"
"time"
)
// Is duration really important ?
type Booking struct {
Start time.Time
Stop time.Time
Duration uint
Workflow string
}
type ScheduledBooking struct {
Bookings []Booking
Mu sync.Mutex
}
func (s Booking) Equals(other Booking) bool {
return s.Workflow == other.Workflow && s.Start == other.Start && s.Stop == other.Stop
}
func (sb *ScheduledBooking) AddSchedule(new_booking Booking){
if(!sb.scheduleAlreadyExists(new_booking)){
sb.Bookings = append(sb.Bookings,new_booking)
logger.Logger.Info().Msg("Updated list schedules : \n " + sb.String())
} else {
// Debug condition : delete once this feature is ready to be implemented
logger.Logger.Debug().Msg("Workflow received not added")
logger.Logger.Debug().Msg("current schedule contains")
for _, booking := range(sb.Bookings){
logger.Logger.Debug().Msg(booking.String())
}
}
}
func (sb *ScheduledBooking) GetListNames()(list_names []string ){
for _, schedule := range(sb.Bookings){
list_names = append(list_names, schedule.Workflow)
}
return
}
func (sb *ScheduledBooking) scheduleAlreadyExists(new_booking Booking) bool {
for _, booking := range(sb.Bookings){
if booking.Equals(new_booking){
return true
}
}
return false
}
func (b *Booking) String() string {
return fmt.Sprintf("{workflow : %s , start_date : %s , stop_date : %s }", b.Workflow, b.Start.Format(time.RFC3339), b.Stop.Format(time.RFC3339))
}
func (sb *ScheduledBooking) String() string {
var str string
for _, booking := range(sb.Bookings){
str += fmt.Sprintf("%s\n", booking.String())
}
return str
}

4
models/template_models.go → models/template.go
View File

@ -31,8 +31,8 @@ type Dag struct {
}
type Template struct {
Name string `yaml:"name"`
Inputs struct {
Name string `yaml:"name"`
Inputs struct {
Parameters []Parameter `yaml:"parameters"`
} `yaml:"inputs,omitempty"`
Container Container `yaml:"container,omitempty"`

17
models/volume.go Normal file
View File

@ -0,0 +1,17 @@
package models
type VolumeClaimTemplate struct {
Metadata struct {
Name string `yaml:"name"`
} `yaml:"metadata"`
Spec VolumeSpec `yaml:"spec"`
}
type VolumeSpec struct {
AccessModes []string `yaml:"accessModes,flow"`
Resources struct {
Requests struct {
Storage string `yaml:"storage"`
} `yaml:"requests"`
} `yaml:"resources"`
}

19
models/volume_models.go
View File

@ -1,19 +0,0 @@
package models
type VolumeClaimTemplate struct {
Metadata struct {
Name string `yaml:"name"`
} `yaml:"metadata"`
Spec VolumeSpec `yaml:"spec"`
}
type VolumeSpec struct {
AccessModes []string `yaml:"accessModes,flow"`
Resources struct {
Requests struct {
Storage string `yaml:"storage"`
} `yaml:"requests"`
} `yaml:"resources"`
}

BIN
oc-monitord Executable file
View File

Binary file not shown.

106
workflow_builder/argo_builder.go
View File

@ -5,14 +5,12 @@
package workflow_builder
import (
"fmt"
"oc-monitor/conf"
"oc-monitor/logger"
. "oc-monitor/models"
. "oc-monitord/models"
"os"
"strings"
"time"
oclib "cloud.o-forge.io/core/oc-lib"
"cloud.o-forge.io/core/oc-lib/models/resource_model"
"cloud.o-forge.io/core/oc-lib/models/resources/workflow/graph"
"github.com/nwtgck/go-fakelish"
@ -46,25 +44,24 @@ func (b *ArgoBuilder) CreateDAG() (string, error) {
b.createDAGstep()
b.createVolumes()
b.Workflow.Spec.Entrypoint = "dag"
b.Workflow.ApiVersion = "argoproj.io/v1alpha1"
b.Workflow.Kind = "Workflow"
random_name := generateWfName()
b.Workflow.Metadata.Name = "oc-monitor-" + random_name
logger := oclib.GetLogger()
yamlified, err := yaml.Marshal(b.Workflow)
if err != nil {
logger.Logger.Error().Msg("Could not transform object to yaml file")
logger.Error().Msg("Could not transform object to yaml file")
return "", err
}
// Give a unique name to each argo file with its timestamp DD:MM:YYYY_hhmmss
current_timestamp := time.Now().Format("02_01_2006_150405")
file_name := random_name + "_" + current_timestamp + ".yml"
workflows_dir := conf.GetConfig().MonitorDir+"/argo_workflows/"
workflows_dir := "./argo_workflows/"
err = os.WriteFile(workflows_dir+file_name, []byte(yamlified), 0660)
if err != nil {
logger.Logger.Error().Msg("Could not write the yaml file")
logger.Error().Msg("Could not write the yaml file")
return "", err
}
@ -72,11 +69,10 @@ func (b *ArgoBuilder) CreateDAG() (string, error) {
}
func (b *ArgoBuilder) createTemplates() {
for _, comp := range b.getProcessings() {
var command string
var args string
var env string
var args string
var env string
comp_res := comp.Processing
@ -107,13 +103,9 @@ func (b *ArgoBuilder) createTemplates() {
}
func (b *ArgoBuilder) createDAGstep() {
new_dag := Dag{}
for _, comp := range b.getProcessings() {
comp_res := comp.Processing
env := getStringValue(comp_res.AbstractResource, "env")
unique_name := getArgoName(comp_res.GetName(), comp.ID)
step := Task{Name: unique_name, Template: unique_name}
@ -124,12 +116,9 @@ func (b *ArgoBuilder) createDAGstep() {
}
// retrieves the name (computing.name-computing.ID)
step.Dependencies = b.getDependency(comp.ID) // Error : we use the component ID instead of the GraphItem ID -> store objects
step.Dependencies = b.getDependency(comp.ID) // Error : we use the component ID instead of the GraphItem ID -> store objects
new_dag.Tasks = append(new_dag.Tasks, step)
}
b.Workflow.Spec.Templates = append(b.Workflow.Spec.Templates, Template{Name: "dag", Dag: new_dag})
}
@ -140,69 +129,36 @@ func (b *ArgoBuilder) createVolumes() {
new_volume.Metadata.Name = "workdir"
new_volume.Spec.AccessModes = []string{"ReadWriteOnce"}
new_volume.Spec.Resources.Requests.Storage = "1Gi"
b.Workflow.Spec.Volumes = append(b.Workflow.Spec.Volumes, new_volume)
}
func (b *ArgoBuilder) getDependency(current_computing_id string) (dependencies []string) {
for _, link := range b.graph.Links {
if !b.IsProcessing(link.Source.ID) || !b.IsProcessing(link.Destination.ID) {
continue
}
source := b.graph.Items[link.Source.ID].Processing
if current_computing_id == link.Destination.ID && source != nil {
dependency_name := getArgoName(source.GetName(), link.Source.ID)
dependencies = append(dependencies, dependency_name)
}
}
return
}
// func (b *ArgoBuilder) componentInBranch(component_id string, branch []string) bool {
// for _, link := range branch {
// if b.graph.Links[link].Source == component_id || b.graph.Links[link].Destination == component_id {
// return true
// }
// }
// return false
// }
// func (b *ArgoBuilder) findPreviousComputing(computing_id string, branch []string, index int) string {
// for i := index; i >= 0 ; i-- {
// previousLink := b.graph.Links[branch[i]]
// if previousLink.Source != computing_id && b.graph.GetComponentType(previousLink.Source) == "computing"{
// name := getArgoName(b.graph.getComponentName(previousLink.Source),previousLink.Source)
// return name
// }
// if previousLink.Destination != computing_id && b.graph.GetComponentType(previousLink.Destination) == "computing"{
// name := getArgoName(b.graph.getComponentName(previousLink.Destination),previousLink.Destination)
// return name
// }
// }
// return ""
// }
func getComputingCommands(user_input string) (list_command []string) {
func getComputingCommands(user_input string) []string {
user_input = removeImageName(user_input)
if len(user_input) == 0 {
return
return []string{}
}
list_command = strings.Split(user_input, " ")
for i := range list_command {
list_command[i] = list_command[i]
}
return
return strings.Split(user_input, " ")
}
func getComputingArgs(user_input []string, command string) (list_args []string) {
if len(user_input) == 0 {
return
}
// quickfix that might need improvement
if strings.Contains(command, "sh -c") {
list_args = append(list_args, strings.Join(user_input, " "))
@ -210,13 +166,12 @@ func getComputingArgs(user_input []string, command string) (list_args []string)
}
list_args = append(list_args, user_input...)
return
}
// Currently implements code to overcome problems in data structure
func getComputingEnvironment(user_input []string) (map_env map[string]string) {
logger := oclib.GetLogger()
is_empty := len(user_input) == 0
is_empty_string := len(user_input) == 1 && user_input[0] == ""
@ -234,7 +189,7 @@ func getComputingEnvironment(user_input []string) (map_env map[string]string) {
new_pair := strings.Split(str, "=")
if len(new_pair) != 2 {
logger.Logger.Error().Msg("Error extracting the environment variable from " + str)
logger.Error().Msg("Error extracting the environment variable from " + str)
panic(0)
}
@ -265,15 +220,6 @@ func getArgoName(raw_name string, component_id string) (formatedName string) {
return
}
func printYAML(data interface{}) {
yamlData, err := yaml.Marshal(data)
if err != nil {
fmt.Printf("Error marshalling YAML: %v\n", err)
return
}
fmt.Println(string(yamlData))
}
func removeImageName(user_input string) string {
// First command is the name of the container for now
if len(strings.Split(user_input, " ")) == 1 {
@ -297,10 +243,18 @@ func (b *ArgoBuilder) getProcessings() (list_computings []graph.GraphItem) {
return
}
func getStringValue(comp resource_model.AbstractResource, key string) string {
if res := comp.GetModelValue(key); res != nil {
return res.(string)
}
return ""
func (b *ArgoBuilder) IsProcessing(id string) bool {
for _, item := range b.graph.Items {
if item.Processing != nil && item.Processing.GetID() == id {
return true
}
}
return false
}
func getStringValue(comp resource_model.AbstractResource, key string) string {
if res := comp.GetModelValue(key); res != nil {
return res.(string)
}
return ""
}

130
workflow_builder/graph.go
View File

@ -4,156 +4,58 @@ import (
"errors"
"fmt"
"oc-monitor/logger"
models "oc-monitor/models"
oclib "cloud.o-forge.io/core/oc-lib"
"cloud.o-forge.io/core/oc-lib/models/resources/workflow/graph"
workflow "cloud.o-forge.io/core/oc-lib/models/workflow"
)
type WorflowDB struct {
workflow_name string // used to test if the graph has been instatiated, private so can only be set by a graph's method
graph *graph.Graph
ws models.HttpQuery
Workflow *workflow.Workflow
}
// Create the obj!ects from the mxgraphxml stored in the workflow given as a parameter
func (w *WorflowDB) LoadFrom(workflow_id string) error {
new_wf, err := w.getWorkflow(workflow_id)
if err != nil {
var err error
if w.Workflow, err = w.getWorkflow(workflow_id); err != nil {
return err
}
w.graph = new_wf.Graph
w.workflow_name = new_wf.Name
return nil
}
// Use oclib to retrieve the graph contained in the workflow referenced
func (w *WorflowDB) getWorkflow( workflow_id string) (workflow *workflow.Workflow, err error) {
func (w *WorflowDB) getWorkflow(workflow_id string) (workflow *workflow.Workflow, err error) {
logger := oclib.GetLogger()
lib_data := oclib.LoadOne(oclib.LibDataEnum(oclib.WORKFLOW),workflow_id)
lib_data := oclib.LoadOne(oclib.LibDataEnum(oclib.WORKFLOW), workflow_id)
if lib_data.Code != 200 {
logger.Logger.Error().Msg("Error loading the graph")
logger.Error().Msg("Error loading the graph")
return workflow, errors.New(lib_data.Err)
}
new_wf := lib_data.ToWorkflow()
if new_wf == nil {
logger.Logger.Error().Msg("WorflowDB object is empty for " + workflow_id )
logger.Error().Msg("WorflowDB object is empty for " + workflow_id)
return workflow, errors.New("WorflowDB can't be empty")
}
return new_wf, nil
}
func (w *WorflowDB) ExportToArgo() (string, error) {
if len(w.workflow_name) == 0 || &w.graph == nil {
return "",fmt.Errorf("can't export a graph that has not been loaded yet")
logger := oclib.GetLogger()
if len(w.Workflow.Name) == 0 || w.Workflow.Graph == nil {
return "", fmt.Errorf("can't export a graph that has not been loaded yet")
}
argo_builder := ArgoBuilder{graph : *w.graph}
argo_builder := ArgoBuilder{graph: *w.Workflow.Graph}
filename, err := argo_builder.CreateDAG()
if err != nil {
logger.Logger.Error().Msg("Could not create the argo file for " + w.workflow_name)
logger.Error().Msg("Could not create the argo file for " + w.Workflow.Name)
return "", err
}
return filename, nil
return filename, nil
}
// TODO implement this function
func (w *WorflowDB) ExportToHelm(id string) error {
return nil
}
// Return if it exists a link where Destination is the same as comp_id
func (w *WorflowDB) isDestination(comp_id string,link_id int) bool {
for i, link := range w.graph.Links{
if(i !=link_id && link.Destination.ID == comp_id){
return true
}
}
return false
}
// Return if it exists a link where Source is the same as comp_id
func (w *WorflowDB) isSource(comp_id string, link_id int) bool {
for i, link := range w.graph.Links{
if(i !=link_id && link.Source.ID == comp_id && !w.isDCLink(i)){
return true
}
}
return false
}
// returns either computing, data or storage
func GetComponentType(component_id string) string {
if libdata := oclib.LoadOne(oclib.LibDataEnum(oclib.PROCESSING_RESOURCE),component_id); libdata.Code == 200{
return "computing"
}
if libdata := oclib.LoadOne(oclib.LibDataEnum(oclib.DATA_RESOURCE),component_id); libdata.Code == 200{
return "data"
}
if libdata := oclib.LoadOne(oclib.LibDataEnum(oclib.STORAGE_RESOURCE),component_id); libdata.Code == 200{
return "storage"
}
if libdata := oclib.LoadOne(oclib.LibDataEnum(oclib.DATACENTER_RESOURCE),component_id); libdata.Code == 200{
return "datacenter"
}
if libdata := oclib.LoadOne(oclib.LibDataEnum(oclib.WORKFLOW_RESOURCE),component_id); libdata.Code == 200{
return "workflow"
}
return ""
}
// Returns a slice of id, in case the link is made of twice the same type of component
func (w *WorflowDB) getComponentByType(compType string, link graph.GraphLink) (ids []string){
if(GetComponentType(link.Source.ID) == compType){
ids = append(ids, link.Source.ID)
}
if(GetComponentType(link.Destination.ID) == compType){
ids = append(ids, link.Destination.ID)
}
return
}
func (w *WorflowDB) isDCLink(link_id int) bool {
link := w.graph.Links[link_id]
dest := w.graph.Items[link.Destination.ID]
dest_id := dest.GetAbstractRessource().GetID()
source := w.graph.Items[link.Source.ID]
source_id := source.GetAbstractRessource().GetID()
return IsDatacenter(dest_id) || IsDatacenter(source_id)
}
func IsDatacenter(id string) bool {
resource := oclib.LoadOne(oclib.LibDataEnum(oclib.DATACENTER_RESOURCE),id)
return resource.Code == 200
}