Installer: dynamically generate open port requests
App config can mark any of the input (int) fields as having a role.
For such fields installer first will make port reservation request to
Port Allocator, which will dynamically allocate and reserve one of the
available ports for the application. Once application is committed to
config repository, installer makes another request to port allocator
to actually open dynamically reserved port in the ingress service.
Added port reservation logic to Port Allocator. Reservation lasts 30
minutes.
Change-Id: Ic8caa0d04459b1a6e8a351e2ca6964ac15c7253d
diff --git a/core/installer/app.go b/core/installer/app.go
index 86447fe..e443948 100644
--- a/core/installer/app.go
+++ b/core/installer/app.go
@@ -82,6 +82,7 @@
type PortForward struct {
Allocator string `json:"allocator"`
+ ReserveAddr string `json:"reservator"`
Protocol string `json:"protocol"`
SourcePort int `json:"sourcePort"`
TargetService string `json:"targetService"`
@@ -437,7 +438,7 @@
networks := CreateNetworks(env)
derived, err := deriveValues(values, a.Schema(), networks)
if err != nil {
- return EnvAppRendered{}, nil
+ return EnvAppRendered{}, err
}
if charts == nil {
charts = make(map[string]helmv2.HelmChartTemplateSpec)
diff --git a/core/installer/app_configs/app_base.cue b/core/installer/app_configs/app_base.cue
index 32fc354..97de21d 100644
--- a/core/installer/app_configs/app_base.cue
+++ b/core/installer/app_configs/app_base.cue
@@ -32,6 +32,7 @@
certificateIssuer: string | *""
domain: string
allocatePortAddr: string
+ reservePortAddr: string
}
#Image: {
@@ -82,6 +83,7 @@
#PortForward: {
allocator: string
+ reservator: string
protocol: "TCP" | "UDP" | *"TCP"
sourcePort: int
targetService: string
diff --git a/core/installer/app_manager.go b/core/installer/app_manager.go
index 612f9a7..4f864e4 100644
--- a/core/installer/app_manager.go
+++ b/core/installer/app_manager.go
@@ -5,13 +5,14 @@
"encoding/json"
"errors"
"fmt"
+ "io"
"io/fs"
"net/http"
"path"
"path/filepath"
"strings"
- "github.com/giolekva/pcloud/core/installer/io"
+ gio "github.com/giolekva/pcloud/core/installer/io"
"github.com/giolekva/pcloud/core/installer/soft"
helmv2 "github.com/fluxcd/helm-controller/api/v2"
@@ -151,9 +152,36 @@
SourcePort int `json:"sourcePort"`
TargetService string `json:"targetService"`
TargetPort int `json:"targetPort"`
+ Secret string `json:"secret"`
}
-func openPorts(ports []PortForward) error {
+type reservePortResp struct {
+ Port int `json:"port"`
+ Secret string `json:"secret"`
+}
+
+func reservePorts(ports map[string]string) (map[string]reservePortResp, error) {
+ ret := map[string]reservePortResp{}
+ for p, reserveAddr := range ports {
+ resp, err := http.Post(reserveAddr, "application/json", nil) // TODO(gio): address
+ if err != nil {
+ return nil, err
+ }
+ if resp.StatusCode != http.StatusOK {
+ var e bytes.Buffer
+ io.Copy(&e, resp.Body)
+ return nil, fmt.Errorf("Could not reserve port: %s", e.String())
+ }
+ var r reservePortResp
+ if err := json.NewDecoder(resp.Body).Decode(&r); err != nil {
+ return nil, err
+ }
+ ret[p] = r
+ }
+ return ret, nil
+}
+
+func openPorts(ports []PortForward, reservations map[string]reservePortResp, allocators map[string]string) error {
for _, p := range ports {
var buf bytes.Buffer
req := allocatePortReq{
@@ -165,7 +193,18 @@
if err := json.NewEncoder(&buf).Encode(req); err != nil {
return err
}
- resp, err := http.Post(p.Allocator, "application/json", &buf)
+ allocator := ""
+ for n, r := range reservations {
+ if p.SourcePort == r.Port {
+ allocator = allocators[n]
+ req.Secret = reservations[n].Secret
+ break
+ }
+ }
+ if allocator == "" {
+ return fmt.Errorf("Could not find allocator for: %d", p.SourcePort)
+ }
+ resp, err := http.Post(allocator, "application/json", &buf)
if err != nil {
return err
}
@@ -183,7 +222,7 @@
kust, err := soft.ReadKustomization(r, kustPath)
if err != nil {
if errors.Is(err, fs.ErrNotExist) {
- k := io.NewKustomization()
+ k := gio.NewKustomization()
kust = &k
} else {
return err
@@ -264,7 +303,7 @@
if err := createKustomizationChain(r, resourcesDir); err != nil {
return "", err
}
- appKust := io.NewKustomization()
+ appKust := gio.NewKustomization()
for name, contents := range resources {
appKust.AddResources(name)
w, err := r.Writer(path.Join(resourcesDir, name))
@@ -293,6 +332,14 @@
values map[string]any,
opts ...InstallOption,
) (ReleaseResources, error) {
+ portFields := findPortFields(app.Schema())
+ fakeReservations := map[string]reservePortResp{}
+ for i, f := range portFields {
+ fakeReservations[f] = reservePortResp{Port: i}
+ }
+ if err := setPortFields(values, fakeReservations); err != nil {
+ return ReleaseResources{}, err
+ }
o := &installOptions{}
for _, i := range opts {
i(o)
@@ -330,6 +377,19 @@
if err != nil {
return ReleaseResources{}, err
}
+ reservators := map[string]string{}
+ allocators := map[string]string{}
+ for _, pf := range rendered.Ports {
+ reservators[portFields[pf.SourcePort]] = pf.ReserveAddr
+ allocators[portFields[pf.SourcePort]] = pf.Allocator
+ }
+ portReservations, err := reservePorts(reservators)
+ if err != nil {
+ return ReleaseResources{}, err
+ }
+ if err := setPortFields(values, portReservations); err != nil {
+ return ReleaseResources{}, err
+ }
imageRegistry := fmt.Sprintf("zot.%s", env.PrivateDomain)
if o.FetchContainerImages {
if err := pullContainerImages(instanceId, rendered.ContainerImages, imageRegistry, namespace, m.jc); err != nil {
@@ -358,7 +418,7 @@
return ReleaseResources{}, err
}
// TODO(gio): add ingress-nginx to release resources
- if err := openPorts(rendered.Ports); err != nil {
+ if err := openPorts(rendered.Ports, portReservations, allocators); err != nil {
return ReleaseResources{}, err
}
return ReleaseResources{
@@ -466,12 +526,14 @@
CertificateIssuer: fmt.Sprintf("%s-public", env.Id),
Domain: env.Domain,
AllocatePortAddr: fmt.Sprintf("http://port-allocator.%s-ingress-public.svc.cluster.local/api/allocate", env.InfraName),
+ ReservePortAddr: fmt.Sprintf("http://port-allocator.%s-ingress-public.svc.cluster.local/api/reserve", env.InfraName),
},
{
Name: "Private",
IngressClass: fmt.Sprintf("%s-ingress-private", env.Id),
Domain: env.PrivateDomain,
AllocatePortAddr: fmt.Sprintf("http://port-allocator.%s-ingress-private.svc.cluster.local/api/allocate", env.Id),
+ ReservePortAddr: fmt.Sprintf("http://port-allocator.%s-ingress-private.svc.cluster.local/api/reserve", env.Id),
},
}
}
@@ -706,3 +768,71 @@
}
return cfg.LocalCharts, nil
}
+
+func findPortFields(scm Schema) []string {
+ switch scm.Kind() {
+ case KindBoolean:
+ return []string{}
+ case KindInt:
+ return []string{}
+ case KindString:
+ return []string{}
+ case KindStruct:
+ ret := []string{}
+ for _, f := range scm.Fields() {
+ for _, p := range findPortFields(f.Schema) {
+ if p == "" {
+ ret = append(ret, f.Name)
+ } else {
+ ret = append(ret, fmt.Sprintf("%s.%s", f.Name, p))
+ }
+ }
+ }
+ return ret
+ case KindNetwork:
+ return []string{}
+ case KindAuth:
+ return []string{}
+ case KindSSHKey:
+ return []string{}
+ case KindNumber:
+ return []string{}
+ case KindArrayString:
+ return []string{}
+ case KindPort:
+ return []string{""}
+ default:
+ panic("MUST NOT REACH!")
+ }
+}
+
+func setPortFields(values map[string]any, ports map[string]reservePortResp) error {
+ for p, r := range ports {
+ if err := setPortField(values, p, r.Port); err != nil {
+ return err
+ }
+ }
+ return nil
+}
+
+func setPortField(values map[string]any, field string, port int) error {
+ f := strings.SplitN(field, ".", 2)
+ if len(f) == 2 {
+ var sub map[string]any
+ if s, ok := values[f[0]]; ok {
+ sub, ok = s.(map[string]any)
+ if !ok {
+ return fmt.Errorf("expected map")
+ }
+ } else {
+ sub = map[string]any{}
+ values[f[0]] = sub
+ }
+ if err := setPortField(sub, f[1], port); err != nil {
+ return err
+ }
+ } else {
+ values[f[0]] = port
+ }
+ return nil
+}
diff --git a/core/installer/app_test.go b/core/installer/app_test.go
index 74dfbdc..5feb719 100644
--- a/core/installer/app_test.go
+++ b/core/installer/app_test.go
@@ -2,13 +2,8 @@
import (
_ "embed"
- "fmt"
"net"
"testing"
-
- "github.com/giolekva/pcloud/core/installer/soft"
-
- "github.com/go-git/go-billy/v5/memfs"
)
var env = EnvConfig{
@@ -260,33 +255,11 @@
for _, r := range rendered.Resources {
t.Log(string(r))
}
- for _, r := range rendered.HelmCharts.Git {
- t.Log(fmt.Sprintf("%+v\n", r))
- }
for _, r := range rendered.Data {
t.Log(string(r))
}
}
-func TestPullGitHelmCharts(t *testing.T) {
- charts := HelmCharts{
- Git: map[string]HelmChartGitRepo{
- "rpuppy": HelmChartGitRepo{
- Address: "https://code.v1.dodo.cloud/pcloud",
- Branch: "main",
- Path: "charts/rpuppy",
- },
- },
- }
- fs := soft.NewBillyRepoFS(memfs.New())
- hf := NewGitHelmFetcher()
- pulled, err := pullHelmCharts(hf, charts, fs, "/helm-charts")
- if err != nil {
- t.Fatal(err)
- }
- fmt.Println(pulled)
-}
-
func TestDNSGateway(t *testing.T) {
contents, err := valuesTmpls.ReadFile("values-tmpl/dns-gateway.cue")
if err != nil {
diff --git a/core/installer/derived.go b/core/installer/derived.go
index f623392..542f70d 100644
--- a/core/installer/derived.go
+++ b/core/installer/derived.go
@@ -19,6 +19,7 @@
CertificateIssuer string `json:"certificateIssuer,omitempty"`
Domain string `json:"domain,omitempty"`
AllocatePortAddr string `json:"allocatePortAddr,omitempty"`
+ ReservePortAddr string `json:"reservePortAddr,omitempty"`
}
type InfraAppInstanceConfig struct {
@@ -81,6 +82,8 @@
ret[k] = v
case KindInt:
ret[k] = v
+ case KindPort:
+ ret[k] = v
case KindArrayString:
a, ok := v.([]string)
if !ok {
@@ -135,6 +138,8 @@
ret[k] = v
case KindInt:
ret[k] = v
+ case KindPort:
+ ret[k] = v
case KindArrayString:
a, ok := v.([]string)
if !ok {
diff --git a/core/installer/go.mod b/core/installer/go.mod
index 73dd5bb..fffd1fa 100644
--- a/core/installer/go.mod
+++ b/core/installer/go.mod
@@ -4,10 +4,6 @@
go 1.22.0
-toolchain go1.22.3
-
-// toolchain go1.21.5
-
require (
cuelang.org/go v0.8.1
github.com/Masterminds/sprig/v3 v3.2.3
diff --git a/core/installer/schema.go b/core/installer/schema.go
index a0acfc3..8ddf73f 100644
--- a/core/installer/schema.go
+++ b/core/installer/schema.go
@@ -2,6 +2,7 @@
import (
"fmt"
+ "strings"
"cuelang.org/go/cue"
"cuelang.org/go/cue/cuecontext"
@@ -19,6 +20,7 @@
KindSSHKey = 6
KindNumber = 4
KindArrayString = 8
+ KindPort = 9
)
type Field struct {
@@ -58,6 +60,7 @@
certificateIssuer: string | *""
domain: string
allocatePortAddr: string
+ reservePortAddr: string
}
value: { %s }
@@ -175,6 +178,11 @@
if nameAttr.Err() == nil {
name = nameAttr.Contents()
}
+ role := ""
+ roleAttr := v.Attribute("role")
+ if roleAttr.Err() == nil {
+ role = strings.ToLower(roleAttr.Contents())
+ }
switch v.IncompleteKind() {
case cue.StringKind:
return basicSchema{name, KindString, false}, nil
@@ -183,7 +191,11 @@
case cue.NumberKind:
return basicSchema{name, KindNumber, false}, nil
case cue.IntKind:
- return basicSchema{name, KindInt, false}, nil
+ if role == "port" {
+ return basicSchema{name, KindPort, true}, nil
+ } else {
+ return basicSchema{name, KindInt, false}, nil
+ }
case cue.ListKind:
return basicSchema{name, KindArrayString, false}, nil
case cue.StructKind:
diff --git a/core/installer/schema_test.go b/core/installer/schema_test.go
new file mode 100644
index 0000000..cca0bd7
--- /dev/null
+++ b/core/installer/schema_test.go
@@ -0,0 +1,36 @@
+package installer
+
+import (
+ "testing"
+)
+
+func TestFindPortFields(t *testing.T) {
+ scm := structSchema{
+ "a",
+ []Field{
+ Field{"x", basicSchema{"x", KindString, false}},
+ Field{"y", basicSchema{"y", KindInt, false}},
+ Field{"z", basicSchema{"z", KindPort, false}},
+ Field{
+ "w",
+ structSchema{
+ "w",
+ []Field{
+ Field{"x", basicSchema{"x", KindString, false}},
+ Field{"y", basicSchema{"y", KindInt, false}},
+ Field{"z", basicSchema{"z", KindPort, false}},
+ },
+ false,
+ },
+ },
+ },
+ false,
+ }
+ p := findPortFields(scm)
+ if len(p) != 2 {
+ t.Fatalf("expected two port fields, %v", p)
+ }
+ if p[0] != "z" || p[1] != "w.z" {
+ t.Fatalf("expected 'z' and 'w.z' port fields, %v", p)
+ }
+}
diff --git a/core/installer/values-tmpl/dodo-app.cue b/core/installer/values-tmpl/dodo-app.cue
index a03566b..238b142 100644
--- a/core/installer/values-tmpl/dodo-app.cue
+++ b/core/installer/values-tmpl/dodo-app.cue
@@ -7,7 +7,7 @@
input: {
network: #Network @name(Network)
subdomain: string @name(Subdomain)
- sshPort: int @name(SSH Port)
+ sshPort: int @name(SSH Port) @role(port)
adminKey: string @name(Admin SSH Public Key)
// TODO(gio): auto generate
@@ -55,6 +55,7 @@
portForward: [#PortForward & {
allocator: input.network.allocatePortAddr
+ reservator: input.network.reservePortAddr
sourcePort: input.sshPort
// TODO(gio): namespace part must be populated by app manager. Otherwise
// third-party app developer might point to a service from different namespace.
diff --git a/core/installer/values-tmpl/gerrit.cue b/core/installer/values-tmpl/gerrit.cue
index d89e0c6..c68e9ac 100644
--- a/core/installer/values-tmpl/gerrit.cue
+++ b/core/installer/values-tmpl/gerrit.cue
@@ -2,7 +2,7 @@
network: #Network @name(Network)
subdomain: string @name(Subdomain)
key: #SSHKey
- sshPort: int @name(SSH Port)
+ sshPort: int @name(SSH Port) @role(port)
}
_domain: "\(input.subdomain).\(input.network.domain)"
@@ -105,6 +105,7 @@
portForward: [#PortForward & {
allocator: input.network.allocatePortAddr
+ reservator: input.network.reservePortAddr
sourcePort: input.sshPort
// TODO(gio): namespace part must be populated by app manager. Otherwise
// third-party app developer might point to a service from different namespace.
diff --git a/core/installer/values-tmpl/soft-serve.cue b/core/installer/values-tmpl/soft-serve.cue
index 5c96771..10da2ac 100644
--- a/core/installer/values-tmpl/soft-serve.cue
+++ b/core/installer/values-tmpl/soft-serve.cue
@@ -1,7 +1,7 @@
input: {
network: #Network @name(Network)
subdomain: string @name(Subdomain)
- sshPort: int @name(SSH Port)
+ sshPort: int @name(SSH Port) @role(port)
adminKey: string @name(Admin SSH Public Key)
}
@@ -46,6 +46,7 @@
portForward: [#PortForward & {
allocator: input.network.allocatePortAddr
+ reservator: input.network.reservePortAddr
sourcePort: input.sshPort
// TODO(gio): namespace part must be populated by app manager. Otherwise
// third-party app developer might point to a service from different namespace.
diff --git a/core/port-allocator/go.mod b/core/port-allocator/go.mod
index c7c5103..4e1db75 100644
--- a/core/port-allocator/go.mod
+++ b/core/port-allocator/go.mod
@@ -2,7 +2,7 @@
replace github.com/giolekva/pcloud/core/installer => /Users/lekva/dev/src/pcloud/core/installer
-go 1.21.5
+go 1.22.0
require (
github.com/giolekva/pcloud/core/installer v0.0.0-20240403111418-e9c05499ec80
diff --git a/core/port-allocator/main.go b/core/port-allocator/main.go
index ac063da..861b1dd 100644
--- a/core/port-allocator/main.go
+++ b/core/port-allocator/main.go
@@ -1,15 +1,19 @@
package main
import (
+ "crypto/rand"
"encoding/json"
"flag"
"fmt"
"io"
"log"
+ "math/big"
"net/http"
"os"
"strconv"
"strings"
+ "sync"
+ "time"
"github.com/giolekva/pcloud/core/installer/soft"
@@ -50,9 +54,11 @@
}
type server struct {
- s *http.Server
- r *http.ServeMux
- client client
+ l sync.Locker
+ s *http.Server
+ r *http.ServeMux
+ client client
+ reserve map[int]string
}
func newServer(port int, client client) *server {
@@ -61,11 +67,12 @@
Addr: fmt.Sprintf(":%d", port),
Handler: r,
}
- return &server{s, r, client}
+ return &server{&sync.Mutex{}, s, r, client, make(map[int]string)}
}
func (s *server) Start() error {
s.r.HandleFunc("/api/allocate", s.handleAllocate)
+ s.r.HandleFunc("/api/reserve", s.handleReserve)
if err := s.s.ListenAndServe(); err != nil && err != http.ErrServerClosed {
return err
}
@@ -81,6 +88,7 @@
SourcePort int `json:"sourcePort"`
TargetService string `json:"targetService"`
TargetPort int `json:"targetPort"`
+ Secret string `json:"secret,omitempty"`
}
func extractAllocateReq(r io.Reader) (allocateReq, error) {
@@ -95,6 +103,11 @@
return req, nil
}
+type reserveResp struct {
+ Port int `json:"port"`
+ Secret string `json:"secret"`
+}
+
func extractPorts(rel map[string]any) (map[string]any, map[string]any, error) {
spec, ok := rel["spec"]
if !ok {
@@ -142,6 +155,24 @@
return nil
}
+const start = 49152
+const end = 65535
+
+func reservePort(pm map[string]struct{}, reserve map[int]string) (int, error) {
+ for i := 0; i < 3; i++ {
+ r, err := rand.Int(rand.Reader, big.NewInt(end-start))
+ if err != nil {
+ return -1, err
+ }
+ p := start + int(r.Int64())
+ ps := strconv.Itoa(p)
+ if _, ok := pm[ps]; !ok {
+ return p, nil
+ }
+ }
+ return -1, fmt.Errorf("could not generate random port")
+}
+
func (s *server) handleAllocate(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPost {
http.Error(w, "only post method is supported", http.StatusBadRequest)
@@ -152,31 +183,36 @@
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
- fmt.Printf("%+v\n", req)
+ if req.Secret != "" {
+ http.Error(w, "secret missing", http.StatusBadRequest)
+ return
+ }
+ s.l.Lock()
+ defer s.l.Unlock()
ingressRel, err := s.client.ReadRelease()
if err != nil {
- fmt.Println(err)
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
- fmt.Printf("%+v\n", ingressRel)
tcp, udp, err := extractPorts(ingressRel)
if err != nil {
- fmt.Println(err)
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
- fmt.Printf("%+v %+v\n", tcp, udp)
+ if val, ok := s.reserve[req.SourcePort]; !ok || val != req.Secret {
+ http.Error(w, "invalid secret", http.StatusBadRequest)
+ return
+ } else {
+ delete(s.reserve, req.SourcePort)
+ }
switch req.Protocol {
case "tcp":
if err := addPort(tcp, req); err != nil {
- fmt.Println(err)
http.Error(w, err.Error(), http.StatusConflict)
return
}
case "udp":
if err := addPort(udp, req); err != nil {
- fmt.Println(err)
http.Error(w, err.Error(), http.StatusConflict)
return
}
@@ -190,6 +226,50 @@
}
}
+func (s *server) handleReserve(w http.ResponseWriter, r *http.Request) {
+ if r.Method != http.MethodPost {
+ http.Error(w, "only post method is supported", http.StatusBadRequest)
+ return
+ }
+ s.l.Lock()
+ defer s.l.Unlock()
+ ingressRel, err := s.client.ReadRelease()
+ if err != nil {
+ http.Error(w, err.Error(), http.StatusInternalServerError)
+ return
+ }
+ tcp, udp, err := extractPorts(ingressRel)
+ if err != nil {
+ http.Error(w, err.Error(), http.StatusInternalServerError)
+ return
+ }
+ var port int
+ used := map[string]struct{}{}
+ for p, _ := range tcp {
+ used[p] = struct{}{}
+ }
+ for p, _ := range udp {
+ used[p] = struct{}{}
+ }
+ if port, err = reservePort(used, s.reserve); err != nil {
+ http.Error(w, err.Error(), http.StatusInternalServerError)
+ return
+ }
+ secret := generateSecret()
+ s.reserve[port] = secret
+ go func() {
+ time.Sleep(30 * time.Minute)
+ s.l.Lock()
+ defer s.l.Unlock()
+ delete(s.reserve, port)
+ }()
+ resp := reserveResp{port, secret}
+ if err := json.NewEncoder(w).Encode(resp); err != nil {
+ http.Error(w, err.Error(), http.StatusInternalServerError)
+ return
+ }
+}
+
// TODO(gio): deduplicate
func createRepoClient(addr string, keyPath string) (soft.RepoIO, error) {
sshKey, err := os.ReadFile(keyPath)
@@ -211,6 +291,11 @@
return soft.NewRepoIO(repo, signer)
}
+func generateSecret() string {
+ // TODO(gio): implement
+ return "foo"
+}
+
func main() {
flag.Parse()
repo, err := createRepoClient(*repoAddr, *sshKey)