From bbe9dcaf2a347c25f66812fbb9baa0d34885c1ef Mon Sep 17 00:00:00 2001 From: restitux Date: Sun, 16 Oct 2022 14:54:49 -0600 Subject: [PATCH] WIP: Replace job execution with runner delegation Server now accepts runner connections via websockets. Jobs will be delegated to these runners. This is currently a WIP, as it is still hardcoded to use the first available runner. --- config/config.go | 6 ++ go.mod | 20 +--- go.sum | 134 ++++++++++++--------------- jobscheduler/jobscheduler.go | 171 +++++++++++++++++++++++++++++++++++ listen/listen.go | 19 +++- main.go | 6 +- runner/runner.go | 112 ----------------------- 7 files changed, 255 insertions(+), 213 deletions(-) create mode 100644 jobscheduler/jobscheduler.go delete mode 100644 runner/runner.go diff --git a/config/config.go b/config/config.go index 2ac0326..8aa6cff 100644 --- a/config/config.go +++ b/config/config.go @@ -16,10 +16,15 @@ type Job struct { Cron *string } +type Runner struct { + Secret string +} + type Config struct { Address string Port int Jobs map[string]Job + Runners map[string]Runner } func GetConfig() (config.Config[Config], error) { @@ -30,6 +35,7 @@ func GetConfig() (config.Config[Config], error) { Address: "127.0.0.1", Port: 45420, Jobs: make(map[string]Job), + Runners: make(map[string]Runner), }, } diff --git a/go.mod b/go.mod index c61c96a..90e1e3b 100644 --- a/go.mod +++ b/go.mod @@ -4,25 +4,13 @@ go 1.19 require ( git.ohea.xyz/golang/config v0.0.0-20220915224621-b9debd233173 - github.com/docker/docker v20.10.18+incompatible github.com/op/go-logging v0.0.0-20160315200505-970db520ece7 + nhooyr.io/websocket v1.8.7 ) require ( - github.com/Microsoft/go-winio v0.5.2 // indirect - github.com/docker/distribution v2.8.1+incompatible // indirect - github.com/docker/go-connections v0.4.0 // indirect - github.com/docker/go-units v0.5.0 // indirect - github.com/gogo/protobuf v1.3.2 // indirect - github.com/moby/term v0.0.0-20220808134915-39b0c02b01ae // indirect - github.com/morikuni/aec v1.0.0 // indirect - github.com/opencontainers/go-digest v1.0.0 // indirect - github.com/opencontainers/image-spec v1.0.2 // indirect + github.com/gobwas/ws v1.1.0 // indirect + github.com/klauspost/compress v1.10.3 // indirect github.com/pelletier/go-toml/v2 v2.0.5 // indirect - github.com/pkg/errors v0.9.1 // indirect - github.com/sirupsen/logrus v1.9.0 // indirect - golang.org/x/net v0.0.0-20201021035429-f5854403a974 // indirect - golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8 // indirect - golang.org/x/time v0.0.0-20220722155302-e5dcc9cfc0b9 // indirect - gotest.tools/v3 v3.3.0 // indirect + golang.org/x/sys v0.0.0-20220728004956-3c1f35247d10 // indirect ) diff --git a/go.sum b/go.sum index 3b57707..e7aaf42 100644 --- a/go.sum +++ b/go.sum @@ -1,100 +1,80 @@ git.ohea.xyz/golang/config v0.0.0-20220915224621-b9debd233173 h1:dhq/W6sa5KkLHVBwwgcNIPWcO4YK2/ecFTTln2W+1n8= git.ohea.xyz/golang/config v0.0.0-20220915224621-b9debd233173/go.mod h1:86PbXJ2WdqQ+3hYqrnv3ukgKNRK9nQfThnlY03FAO0g= -github.com/Azure/go-ansiterm v0.0.0-20210617225240-d185dfc1b5a1 h1:UQHMgLO+TxOElx5B5HZ4hJQsoJ/PvUvKRhJHDQXO8P8= -github.com/Azure/go-ansiterm v0.0.0-20210617225240-d185dfc1b5a1/go.mod h1:xomTg63KZ2rFqZQzSB4Vz2SUXa1BpHTVz9L5PTmPC4E= -github.com/Microsoft/go-winio v0.5.2 h1:a9IhgEQBCUEk6QCdml9CiJGhAws+YwffDHEMp1VMrpA= -github.com/Microsoft/go-winio v0.5.2/go.mod h1:WpS1mjBmmwHBEWmogvA2mj8546UReBk4v8QkMxJ6pZY= -github.com/creack/pty v1.1.11/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= -github.com/docker/distribution v2.8.1+incompatible h1:Q50tZOPR6T/hjNsyc9g8/syEs6bk8XXApsHjKukMl68= -github.com/docker/distribution v2.8.1+incompatible/go.mod h1:J2gT2udsDAN96Uj4KfcMRqY0/ypR+oyYUYmja8H+y+w= -github.com/docker/docker v20.10.18+incompatible h1:SN84VYXTBNGn92T/QwIRPlum9zfemfitN7pbsp26WSc= -github.com/docker/docker v20.10.18+incompatible/go.mod h1:eEKB0N0r5NX/I1kEveEz05bcu8tLC/8azJZsviup8Sk= -github.com/docker/go-connections v0.4.0 h1:El9xVISelRB7BuFusrZozjnkIM5YnzCViNKohAFqRJQ= -github.com/docker/go-connections v0.4.0/go.mod h1:Gbd7IOopHjR8Iph03tsViu4nIes5XhDvyHbTtUxmeec= -github.com/docker/go-units v0.5.0 h1:69rxXcBk27SvSaaxTtLh/8llcHD8vYHT7WSdRZ/jvr4= -github.com/docker/go-units v0.5.0/go.mod h1:fgPhTUdO+D/Jk86RDLlptpiXQzgHJF7gydDDbaIK4Dk= -github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q= -github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q= -github.com/google/go-cmp v0.3.0/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU= +github.com/gin-contrib/sse v0.1.0 h1:Y/yl/+YNO8GZSjAhjMsSuLt29uWRFHdHYUb5lYOV9qE= +github.com/gin-contrib/sse v0.1.0/go.mod h1:RHrZQHXnP2xjPF+u1gW/2HnVO7nvIa9PG3Gm+fLHvGI= +github.com/gin-gonic/gin v1.6.3 h1:ahKqKTFpO5KTPHxWZjEdPScmYaGtLo8Y4DMHoEsnp14= +github.com/gin-gonic/gin v1.6.3/go.mod h1:75u5sXoLsGZoRN5Sgbi1eraJ4GU3++wFwWzhwvtwp4M= +github.com/go-playground/assert/v2 v2.0.1/go.mod h1:VDjEfimB/XKnb+ZQfWdccd7VUvScMdVu0Titje2rxJ4= +github.com/go-playground/locales v0.13.0 h1:HyWk6mgj5qFqCT5fjGBuRArbVDfE4hi8+e8ceBS/t7Q= +github.com/go-playground/locales v0.13.0/go.mod h1:taPMhCMXrRLJO55olJkUXHZBHCxTMfnGwq/HNwmWNS8= +github.com/go-playground/universal-translator v0.17.0 h1:icxd5fm+REJzpZx7ZfpaD876Lmtgy7VtROAbHHXk8no= +github.com/go-playground/universal-translator v0.17.0/go.mod h1:UkSxE5sNxxRwHyU+Scu5vgOQjsIJAF8j9muTVoKLVtA= +github.com/go-playground/validator/v10 v10.2.0 h1:KgJ0snyC2R9VXYN2rneOtQcw5aHQB1Vv0sFl1UcHBOY= +github.com/go-playground/validator/v10 v10.2.0/go.mod h1:uOYAAleCW8F/7oMFd6aG0GOhaH6EGOAJShg8Id5JGkI= +github.com/gobwas/httphead v0.0.0-20180130184737-2c6c146eadee/go.mod h1:L0fX3K22YWvt/FAX9NnzrNzcI4wNYi9Yku4O0LKYflo= +github.com/gobwas/httphead v0.1.0 h1:exrUm0f4YX0L7EBwZHuCF4GDp8aJfVeBrlLQrs6NqWU= +github.com/gobwas/httphead v0.1.0/go.mod h1:O/RXo79gxV8G+RqlR/otEwx4Q36zl9rqC5u12GKvMCM= +github.com/gobwas/pool v0.2.0/go.mod h1:q8bcK0KcYlCgd9e7WYLm9LpyS+YeLd8JVDW6WezmKEw= +github.com/gobwas/pool v0.2.1 h1:xfeeEhW7pwmX8nuLVlqbzVc7udMDrwetjEv+TZIz1og= +github.com/gobwas/pool v0.2.1/go.mod h1:q8bcK0KcYlCgd9e7WYLm9LpyS+YeLd8JVDW6WezmKEw= +github.com/gobwas/ws v1.0.2/go.mod h1:szmBTxLgaFppYjEmNtny/v3w89xOydFnnZMcgRRu/EM= +github.com/gobwas/ws v1.1.0 h1:7RFti/xnNkMJnrK7D1yQ/iCIB5OrrY/54/H930kIbHA= +github.com/gobwas/ws v1.1.0/go.mod h1:nzvNcVha5eUziGrbxFCo6qFIojQHjJV5cLYIbezhfL0= +github.com/golang/protobuf v1.3.3/go.mod h1:vzj43D7+SQXF/4pzW/hwtAqwc6iTitCiVSaWz5lYuqw= +github.com/golang/protobuf v1.3.5 h1:F768QJ1E9tib+q5Sc8MkdJi1RxLTbRcTf8LJV56aRls= +github.com/golang/protobuf v1.3.5/go.mod h1:6O5/vntMXwX2lRkT1hjjk0nAC1IDOTvTlVgjlRvqsdk= +github.com/google/go-cmp v0.4.0 h1:xsAVV57WRhGj6kEIi8ReJzQlHHqcBYCElAvkovg3B/4= github.com/google/go-cmp v0.4.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= -github.com/google/go-cmp v0.5.5 h1:Khx7svrCpmxxtHBq5j2mp/xVjsi8hQMfNLvJFAlrGgU= -github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= -github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8= -github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck= -github.com/moby/term v0.0.0-20220808134915-39b0c02b01ae h1:O4SWKdcHVCvYqyDV+9CJA1fcDN2L11Bule0iFy3YlAI= -github.com/moby/term v0.0.0-20220808134915-39b0c02b01ae/go.mod h1:E2VnQOmVuvZB6UYnnDB0qG5Nq/1tD9acaOpo6xmt0Kw= -github.com/morikuni/aec v1.0.0 h1:nP9CBfwrvYnBRgY6qfDQkygYDmYwOilePFkwzv4dU8A= -github.com/morikuni/aec v1.0.0/go.mod h1:BbKIizmSmc5MMPqRYbxO4ZU0S0+P200+tUnFx7PXmsc= +github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg= +github.com/gorilla/websocket v1.4.1 h1:q7AeDBpnBk8AogcD4DSag/Ukw/KV+YhzLj2bP5HvKCM= +github.com/gorilla/websocket v1.4.1/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE= +github.com/json-iterator/go v1.1.9 h1:9yzud/Ht36ygwatGx56VwCZtlI/2AD15T1X2sjSuGns= +github.com/json-iterator/go v1.1.9/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/uOdHXbAo4= +github.com/klauspost/compress v1.10.3 h1:OP96hzwJVBIHYU52pVTI6CczrxPvrGfgqF9N5eTO0Q8= +github.com/klauspost/compress v1.10.3/go.mod h1:aoV0uJVorq1K+umq18yTdKaF57EivdYsUV+/s2qKfXs= +github.com/leodido/go-urn v1.2.0 h1:hpXL4XnriNwQ/ABnpepYM/1vCLWNDfUNts8dX3xTG6Y= +github.com/leodido/go-urn v1.2.0/go.mod h1:+8+nEpDfqqsY+g338gtMEUOtuK+4dEMhiQEgxpxOKII= +github.com/mattn/go-isatty v0.0.12 h1:wuysRhFDzyxgEmMf5xjvJ2M9dZoWAXNNr5LSBS7uHXY= +github.com/mattn/go-isatty v0.0.12/go.mod h1:cbi8OIDigv2wuxKPP5vlRcQ1OAZbq2CE4Kysco4FUpU= +github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421 h1:ZqeYNhU3OHLH3mGKHDcjJRFFRrJa6eAM5H+CtDdOsPc= +github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q= +github.com/modern-go/reflect2 v0.0.0-20180701023420-4b7aa43c6742 h1:Esafd1046DLDQ0W1YjYsBW+p8U2u7vzgW2SQVmlNazg= +github.com/modern-go/reflect2 v0.0.0-20180701023420-4b7aa43c6742/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3RllmbCylyMrvgv0= github.com/op/go-logging v0.0.0-20160315200505-970db520ece7 h1:lDH9UUVJtmYCjyT0CI4q8xvlXPxeZ0gYCVvWbmPlp88= github.com/op/go-logging v0.0.0-20160315200505-970db520ece7/go.mod h1:HzydrMdWErDVzsI23lYNej1Htcns9BCg93Dk0bBINWk= -github.com/opencontainers/go-digest v1.0.0 h1:apOUWs51W5PlhuyGyz9FCeeBIOUDA/6nW8Oi/yOhh5U= -github.com/opencontainers/go-digest v1.0.0/go.mod h1:0JzlMkj0TRzQZfJkVvzbP0HBR3IKzErnv2BNG4W4MAM= -github.com/opencontainers/image-spec v1.0.2 h1:9yCKha/T5XdGtO0q9Q9a6T5NUCsTn/DrBg0D7ufOcFM= -github.com/opencontainers/image-spec v1.0.2/go.mod h1:BtxoFyWECRxE4U/7sNtV5W15zMzWCbyJoFRP3s7yZA0= github.com/pelletier/go-toml/v2 v2.0.5 h1:ipoSadvV8oGUjnUbMub59IDPPwfxF694nG/jwbMiyQg= github.com/pelletier/go-toml/v2 v2.0.5/go.mod h1:OMHamSCAODeSsVrwwvcJOaoN0LIUIaFVNZzmWyNfXas= -github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= -github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4= -github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= -github.com/sirupsen/logrus v1.7.0/go.mod h1:yWOB1SBYBC5VeMP7gHvWumXLIWorT60ONWic61uBYv0= -github.com/sirupsen/logrus v1.9.0 h1:trlNQbNUG3OdDrDil03MCb1H2o9nJ1x4/5LYw7byDE0= -github.com/sirupsen/logrus v1.9.0/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ= -github.com/spf13/pflag v1.0.3/go.mod h1:DYY7MBk1bdzusC3SYhjObp+wFpr4gzcvqqNjLnInEg4= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw= -github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs= -github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= +github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= +github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4= github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= github.com/stretchr/testify v1.8.0 h1:pSgiaMZlXftHpm5L7V1+rVB+AZJydKsMxsQBIJw4PKk= github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU= -github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= -github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= -golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= -golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI= -golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto= -golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= -golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= -golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= -golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= -golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= -golang.org/x/net v0.0.0-20200226121028-0de0cce0169b/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= -golang.org/x/net v0.0.0-20201021035429-f5854403a974 h1:IX6qOQeG5uLjB/hjjwjedwfjND0hgjPMMyO1RoIXQNI= -golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU= -golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= -golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= -golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= -golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= -golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= -golang.org/x/sys v0.0.0-20191026070338-33540a1f6037/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= -golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= -golang.org/x/sys v0.0.0-20210119212857-b64e53b001e4/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= -golang.org/x/sys v0.0.0-20210124154548-22da62e12c0c/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= -golang.org/x/sys v0.0.0-20210616094352-59db8d763f22/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= -golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8 h1:0A+M6Uqn+Eje4kHMK80dtF3JCXC4ykBgQG4Fe06QRhQ= -golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= -golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= -golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= -golang.org/x/time v0.0.0-20220722155302-e5dcc9cfc0b9 h1:ftMN5LMiBFjbzleLqtoBZk7KdJwhuybIU+FckUHgoyQ= -golang.org/x/time v0.0.0-20220722155302-e5dcc9cfc0b9/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ= +github.com/ugorji/go v1.1.7 h1:/68gy2h+1mWMrwZFeD1kQialdSzAb432dtpeJ42ovdo= +github.com/ugorji/go v1.1.7/go.mod h1:kZn38zHttfInRq0xu/PH0az30d+z6vm202qpg1oXVMw= +github.com/ugorji/go/codec v1.1.7 h1:2SvQaVZ1ouYrrKKwoSk2pzd4A9evlKJb9oTL+OaLUSs= +github.com/ugorji/go/codec v1.1.7/go.mod h1:Ax+UKWsSmolVDwsd+7N3ZtXu+yMGCf907BLYF3GoBXY= +golang.org/x/sys v0.0.0-20200116001909-b77594299b42/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20201207223542-d4d67f95c62d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20220728004956-3c1f35247d10 h1:WIoqL4EROvwiPdUtaip4VcDdpZ4kha7wBWZrbVKCIZg= +golang.org/x/sys v0.0.0-20220728004956-3c1f35247d10/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk= +golang.org/x/time v0.0.0-20191024005414-555d28b269f0/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ= golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= -golang.org/x/tools v0.0.0-20190624222133-a101b041ded4/go.mod h1:/rFqwRUd4F7ZHNgwSSTFct+R/Kf4OFW1sUzUTQQTgfc= -golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= -golang.org/x/tools v0.0.0-20200619180055-7c47624df98f/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE= -golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA= -golang.org/x/tools v0.1.0/go.mod h1:xkSsbof2nBLbhDlRMhhhyNLN/zl3eTqcnHD5viDpcZ0= -golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= -golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4= golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= -golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= +gopkg.in/yaml.v2 v2.2.8 h1:obN1ZagJSUGI0Ek/LBmuj4SNLPfIny3KsKFopxRdj10= +gopkg.in/yaml.v2 v2.2.8/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= -gotest.tools/v3 v3.0.2/go.mod h1:3SzNCllyD9/Y+b5r9JIKQ474KzkZyqLqEfYqMsX94Bk= -gotest.tools/v3 v3.3.0 h1:MfDY1b1/0xN1CyMlQDac0ziEy9zJQd9CXBRRDHw2jJo= -gotest.tools/v3 v3.3.0/go.mod h1:Mcr9QNxkg0uMvy/YElmo4SpXgJKWgQvYrT7Kw5RzJ1A= +nhooyr.io/websocket v1.8.7 h1:usjR2uOr/zjjkVMy0lW+PPohFok7PCow5sDjLgX4P4g= +nhooyr.io/websocket v1.8.7/go.mod h1:B70DZP8IakI65RVQ51MsWP/8jndNma26DVA/nFSCgW0= diff --git a/jobscheduler/jobscheduler.go b/jobscheduler/jobscheduler.go new file mode 100644 index 0000000..92d77e1 --- /dev/null +++ b/jobscheduler/jobscheduler.go @@ -0,0 +1,171 @@ +package jobscheduler + +import ( + "context" + "time" + + "git.ohea.xyz/cursorius/server/config" + "github.com/op/go-logging" + "nhooyr.io/websocket" + "nhooyr.io/websocket/wsjson" +) + +var log = logging.MustGetLogger("cursorius-server") + +type RunnerRegistration struct { + Secret string + Id string + Tags []string + conn *websocket.Conn +} + +type RunnerData struct { + msgType websocket.MessageType + data []byte +} + +type Runner struct { + id string + tags []string + conn *websocket.Conn + receiveChan chan RunnerData + running bool +} + +type jobScheduler struct { + runCh chan Run + registerCh chan RunnerRegistration + connectedRunners []Runner + configuredRunners map[string]config.Runner + jobs map[string]config.Job +} + +type Run struct { + Name string +} + +type runnerJob struct { + URL *string + Folder *string +} + +func runJobScheduler(j jobScheduler) { + for { + select { + case run := <-j.runCh: + log.Infof("Got run: %v", run) + if job, exists := j.jobs[run.Name]; exists { + log.Debugf("Finding runner for job \"%v\"", run.Name) + rJ := runnerJob{ + URL: job.URL, + Folder: job.Folder, + } + launched := false + for i, runner := range j.connectedRunners { + // don't send job to runner that is already occupied + if !runner.running { + // don't send job to runnenr with closed receiveChan (is defunct) + // there should never be messages to read on an inactive runner, + // so we aren't losing any data here + select { + case <-runner.receiveChan: + // if the receive channel is closed, swap delete the runner as it's defunct + j.connectedRunners[i] = j.connectedRunners[len(j.connectedRunners)-1] + j.connectedRunners = j.connectedRunners[:len(j.connectedRunners)-1] + default: + err := wsjson.Write(context.Background(), runner.conn, rJ) + if err != nil { + log.Debugf("Could not launch run: %v", err) + } else { + log.Infof("Launched run for job %v on runner %v", run.Name, runner.id) + launched = true + runner.running = true + break + } + } + } else { + log.Debugf("Skipping runner %v, as runner is activly running another job", runner.id) + } + } + if !launched { + errorMsg := "could not find valid runner" + if len(j.connectedRunners) == 0 { + errorMsg = "no connected runners" + } + log.Errorf("Could not launch run for job \"%v\": %v", run.Name, errorMsg) + } + } else { + log.Errorf("No configured job with name %v", run.Name) + } + case registration := <-j.registerCh: + log.Debugf("New runner appeared with id: %v and secret: %v", registration.Id, registration.Secret) + if configuredRunner, doesExist := j.configuredRunners[registration.Id]; doesExist { + if configuredRunner.Secret == registration.Secret { + log.Infof("Registering runner \"%v\" with tags %v", registration.Id, registration.Tags) + runner := Runner{ + id: registration.Id, + tags: registration.Tags, + conn: registration.conn, + receiveChan: make(chan RunnerData), + running: false, + } + j.connectedRunners = append(j.connectedRunners, runner) + // start goroutine to call Read function on websocket connection + // this is required to keep the connection functioning + go func() { + for { + msgType, data, err := registration.conn.Read(context.Background()) + if err != nil { + // TODO: this is still racy, since a job could be emitted between the + // connection returning an err and the channel closing + close(runner.receiveChan) + log.Errorf("Could not read from connection: %v", err) + log.Noticef("Deregistering runner with id: %v", runner.id) + + return + } else { + log.Debugf("%v: %v", msgType, data) + runner.receiveChan <- RunnerData{msgType: msgType, data: data} + } + } + }() + + } else { + log.Errorf("Disconnecting runner with id: %v and invalid secret: %v", registration.Id, registration.Secret) + registration.conn.Close(websocket.StatusNormalClosure, "registration invalid") + } + } else { + log.Errorf("Disconnecting runner with invalid id: %v", registration.Id) + registration.conn.Close(websocket.StatusNormalClosure, "registration invalid") + } + } + } +} + +func StartJobScheduler(jobs map[string]config.Job, configuredRunners map[string]config.Runner) (chan Run, chan RunnerRegistration, error) { + scheduler := jobScheduler{ + runCh: make(chan Run), + registerCh: make(chan RunnerRegistration), + connectedRunners: make([]Runner, 0), + configuredRunners: configuredRunners, + jobs: jobs, + } + + go runJobScheduler(scheduler) + + return scheduler.runCh, scheduler.registerCh, nil +} + +func RegisterRunner(conn *websocket.Conn, registerCh chan RunnerRegistration) { + ctx, cancel := context.WithTimeout(context.Background(), time.Second*10) + defer cancel() + + var registration RunnerRegistration + registration.conn = conn + err := wsjson.Read(ctx, conn, ®istration) + if err != nil { + log.Errorf("Could not read data from websocket connection: %v", err) + return + } + registerCh <- registration +} diff --git a/listen/listen.go b/listen/listen.go index 593a369..91531d3 100644 --- a/listen/listen.go +++ b/listen/listen.go @@ -2,28 +2,37 @@ package listen import ( "fmt" - "git.ohea.xyz/cursorius/server/runner" + "git.ohea.xyz/cursorius/server/jobscheduler" "github.com/op/go-logging" "net/http" + "nhooyr.io/websocket" ) var log = logging.MustGetLogger("cursorius-server") -func setupHTTPServer(runnerCh chan runner.Run) { +func setupHTTPServer(runCh chan jobscheduler.Run, registerCh chan jobscheduler.RunnerRegistration) { http.HandleFunc("/webhook", func(w http.ResponseWriter, r *http.Request) { switch r.Method { case "POST": log.Info("Got webhook") - runnerCh <- runner.Run{Name: "test-local"} + runCh <- jobscheduler.Run{Name: "test"} default: log.Errorf("Got request with method \"%v\", ignoring...", r.Method) } }) + http.HandleFunc("/runner", func(w http.ResponseWriter, r *http.Request) { + conn, err := websocket.Accept(w, r, nil) + if err != nil { + log.Errorf("Could not upgrade runner connection to websocket: %v", err) + return + } + go jobscheduler.RegisterRunner(conn, registerCh) + }) } -func Listen(address string, port int, runnerCh chan runner.Run) { +func Listen(address string, port int, runCh chan jobscheduler.Run, registerCh chan jobscheduler.RunnerRegistration) { - setupHTTPServer(runnerCh) + setupHTTPServer(runCh, registerCh) connect_string := fmt.Sprintf("%v:%v", address, port) log.Noticef("Launching HTTP server on %v\n", connect_string) diff --git a/main.go b/main.go index 7d98876..f0eeda3 100644 --- a/main.go +++ b/main.go @@ -2,8 +2,8 @@ package main import ( "git.ohea.xyz/cursorius/server/config" + "git.ohea.xyz/cursorius/server/jobscheduler" "git.ohea.xyz/cursorius/server/listen" - "git.ohea.xyz/cursorius/server/runner" "github.com/op/go-logging" "os" ) @@ -29,10 +29,10 @@ func main() { log.Errorf("Could not get configuration: %v", err) } - ch, err := runner.StartRunner(configData.Config.Jobs) + runCh, registerCh, err := jobscheduler.StartJobScheduler(configData.Config.Jobs, configData.Config.Runners) if err != nil { log.Errorf("Could not start runner: %v", err) } - listen.Listen(configData.Config.Address, configData.Config.Port, ch) + listen.Listen(configData.Config.Address, configData.Config.Port, runCh, registerCh) } diff --git a/runner/runner.go b/runner/runner.go deleted file mode 100644 index 46e6e98..0000000 --- a/runner/runner.go +++ /dev/null @@ -1,112 +0,0 @@ -package runner - -import ( - "bytes" - "context" - "fmt" - "os" - - "git.ohea.xyz/cursorius/server/config" - "github.com/docker/docker/api/types" - "github.com/docker/docker/api/types/container" - "github.com/docker/docker/client" - "github.com/docker/docker/pkg/stdcopy" - "github.com/op/go-logging" -) - -var log = logging.MustGetLogger("cursorius-server") - -type Run struct { - Name string -} - -func runJob(job config.Job) { - if job.Folder != nil { - log.Debugf("Job configured with folder \"%v\"", *job.Folder) - - cli, err := client.NewClientWithOpts(client.FromEnv) - if err != nil { - log.Errorf("Could not create docker clien: %va", err) - return - } - - hostname, err := os.Hostname() - if err != nil { - log.Errorf("Could not get hostname: %v", err) - return - } - - ctx := context.Background() - - resp, err := cli.ContainerCreate(ctx, - &container.Config{ - Image: "cursorius:latest", - Cmd: []string{"/launcher.sh"}, - Tty: false, - Env: []string{fmt.Sprintf("CURSORIUS_SRC_DIR=%s", *job.Folder)}, - }, - &container.HostConfig{ - VolumesFrom: []string{hostname}, - }, - nil, nil, "", - ) - if err != nil { - log.Errorf("Could not create container: %v", err) - return - } - - if err := cli.ContainerStart(ctx, resp.ID, types.ContainerStartOptions{}); err != nil { - log.Errorf("Could not start container: %v", err) - return - } - statusCh, errCh := cli.ContainerWait(ctx, resp.ID, container.WaitConditionNotRunning) - select { - case err := <-errCh: - if err != nil { - log.Errorf("Container returned error: %v", err) - return - } - case retCode := <-statusCh: - log.Debugf("Container finished running with return code: %v", retCode) - } - - out, err := cli.ContainerLogs(ctx, resp.ID, types.ContainerLogsOptions{ShowStdout: true, ShowStderr: true}) - if err != nil { - log.Errorf("Could not get container logs: %v", err) - return - } - - var stdOut bytes.Buffer - var stdErr bytes.Buffer - - stdcopy.StdCopy(&stdOut, &stdErr, out) - - log.Debugf("%s", stdOut.Bytes()) - log.Debugf("%s", stdErr.Bytes()) - - } else if job.URL != nil { - log.Debugf("Job configured with URL \"%v\"", *job.URL) - } -} - -func runnerListen(ch chan Run, jobs map[string]config.Job) { - for { - run := <-ch - log.Debugf("Got Run: %v", run) - job, exists := jobs[run.Name] - if exists { - log.Infof("Launching run for job %v", run.Name) - go runJob(job) - } else { - log.Errorf("No configured job with name %v", run.Name) - } - } -} - -func StartRunner(jobs map[string]config.Job) (chan Run, error) { - ch := make(chan Run) - - go runnerListen(ch, jobs) - - return ch, nil -}