-
Notifications
You must be signed in to change notification settings - Fork 5
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Call remove node endpoint on machine deletion
- Loading branch information
1 parent
4f3a678
commit 862748d
Showing
5 changed files
with
412 additions
and
0 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,117 @@ | ||
| package clusteragent | ||
|
|
||
| import ( | ||
| "bytes" | ||
| "context" | ||
| "crypto/tls" | ||
| "encoding/json" | ||
| "errors" | ||
| "fmt" | ||
| "net/http" | ||
| "time" | ||
|
|
||
| "k8s.io/apimachinery/pkg/util/sets" | ||
| clusterv1 "sigs.k8s.io/cluster-api/api/v1beta1" | ||
| ) | ||
|
|
||
| const defaultClusterAgentPort = "25000" | ||
|
|
||
| // Options should be used when initializing a new client. | ||
| type Options struct { | ||
| // IgnoreNodeIPs is a set of ignored IPs that we don't want to pick for the cluster agent endpoint. | ||
| IgnoreNodeIPs sets.String | ||
| // Port overwrites the default cluster agent port to connect. | ||
| Port string | ||
| // InsecureSkipVerify skips the verification of the server's certificate chain and host name. | ||
| InsecureSkipVerify bool | ||
| } | ||
|
|
||
| type Client struct { | ||
| ip, port string | ||
| client *http.Client | ||
| } | ||
|
|
||
| // NewClient picks an IP from one of the given machines and creates a new client for the cluster agent | ||
| // with that IP. | ||
| func NewClient(machines []clusterv1.Machine, opts Options) (*Client, error) { | ||
| var ip string | ||
| for _, m := range machines { | ||
| for _, addr := range m.Status.Addresses { | ||
| if !opts.IgnoreNodeIPs.Has(addr.Address) { | ||
| ip = addr.Address | ||
| break | ||
| } | ||
| } | ||
| } | ||
|
|
||
| if ip == "" { | ||
| return nil, errors.New("failed to find an IP for cluster agent") | ||
| } | ||
|
|
||
| port := defaultClusterAgentPort | ||
| if opts.Port != "" { | ||
| port = opts.Port | ||
| } | ||
|
|
||
| transport := &http.Transport{ | ||
| TLSClientConfig: &tls.Config{ | ||
| InsecureSkipVerify: opts.InsecureSkipVerify, | ||
| }, | ||
| } | ||
|
|
||
| return &Client{ | ||
| ip: ip, | ||
| port: port, | ||
| client: &http.Client{ | ||
| Timeout: 30 * time.Second, | ||
| Transport: transport, | ||
| }, | ||
| }, nil | ||
| } | ||
|
|
||
| func (c *Client) Endpoint() string { | ||
| return fmt.Sprintf("https://%s:%s", c.ip, c.port) | ||
| } | ||
|
|
||
| // Do makes a request to the given endpoint with the given method. It marshals the request and unmarshals | ||
| // server response body if the provided response is not nil. | ||
| // The endpoint should _not_ have a leading slash. | ||
| func (c *Client) Do(ctx context.Context, method, endpoint string, request any, response any) error { | ||
| url := fmt.Sprintf("https://%s:%s/%s", c.ip, c.port, endpoint) | ||
|
|
||
| requestBody, err := json.Marshal(request) | ||
| if err != nil { | ||
| return fmt.Errorf("failed to prepare worker info request: %w", err) | ||
| } | ||
|
|
||
| req, err := http.NewRequestWithContext(ctx, method, url, bytes.NewBuffer(requestBody)) | ||
| if err != nil { | ||
| return fmt.Errorf("failed to create request: %w", err) | ||
| } | ||
|
|
||
| res, err := c.client.Do(req) | ||
| if err != nil { | ||
| return fmt.Errorf("failed to call cluster agent: %w", err) | ||
| } | ||
| defer res.Body.Close() | ||
|
|
||
| if res.StatusCode != http.StatusOK { | ||
| // NOTE(hue): Marshal and print any response that we got since it might contain valuable information | ||
| // on why the request failed. | ||
| // Ignore JSON errors to prevent unnecessarily complicated error handling. | ||
| anyResp := make(map[string]any) | ||
| _ = json.NewDecoder(res.Body).Decode(&anyResp) | ||
| b, _ := json.Marshal(anyResp) | ||
| resStr := string(b) | ||
|
|
||
| return fmt.Errorf("HTTP request to cluster agent failed with status code %d, got response: %q", res.StatusCode, resStr) | ||
| } | ||
|
|
||
| if response != nil { | ||
| if err := json.NewDecoder(res.Body).Decode(response); err != nil { | ||
| return fmt.Errorf("failed to decode response: %w", err) | ||
| } | ||
| } | ||
|
|
||
| return nil | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,184 @@ | ||
| package clusteragent_test | ||
|
|
||
| import ( | ||
| "context" | ||
| "encoding/json" | ||
| "fmt" | ||
| "net" | ||
| "net/http" | ||
| "net/http/httptest" | ||
| "strings" | ||
| "testing" | ||
|
|
||
| . "github.com/onsi/gomega" | ||
| "k8s.io/apimachinery/pkg/util/sets" | ||
|
|
||
| "github.com/canonical/cluster-api-control-plane-provider-microk8s/pkg/clusteragent" | ||
| clusterv1 "sigs.k8s.io/cluster-api/api/v1beta1" | ||
| ) | ||
|
|
||
| func TestClient(t *testing.T) { | ||
| t.Run("CanNotFindAddress", func(t *testing.T) { | ||
| g := NewWithT(t) | ||
|
|
||
| // Machines don't have any addresses. | ||
| machines := []clusterv1.Machine{{}, {}} | ||
| _, err := clusteragent.NewClient(machines, clusteragent.Options{}) | ||
|
|
||
| g.Expect(err).To(HaveOccurred()) | ||
|
|
||
| // The only machine is the ignored one. | ||
| addr := "1.1.1.1" | ||
| machines = []clusterv1.Machine{ | ||
| { | ||
| Status: clusterv1.MachineStatus{ | ||
| Addresses: clusterv1.MachineAddresses{ | ||
| { | ||
| Address: addr, | ||
| }, | ||
| }, | ||
| }, | ||
| }, | ||
| } | ||
| _, err = clusteragent.NewClient(machines, clusteragent.Options{IgnoreNodeIPs: sets.NewString(addr)}) | ||
|
|
||
| g.Expect(err).To(HaveOccurred()) | ||
| }) | ||
|
|
||
| t.Run("CorrectEndpoint", func(t *testing.T) { | ||
| g := NewWithT(t) | ||
|
|
||
| port := "30000" | ||
| firstAddr := "1.1.1.1" | ||
| secondAddr := "2.2.2.2" | ||
| thirdAddr := "3.3.3.3" | ||
| ignoreAddr := "8.8.8.8" | ||
| machines := []clusterv1.Machine{ | ||
| { | ||
| Status: clusterv1.MachineStatus{ | ||
| Addresses: clusterv1.MachineAddresses{ | ||
| { | ||
| Address: firstAddr, | ||
| }, | ||
| }, | ||
| }, | ||
| }, | ||
| { | ||
| Status: clusterv1.MachineStatus{ | ||
| Addresses: clusterv1.MachineAddresses{ | ||
| { | ||
| Address: secondAddr, | ||
| }, | ||
| }, | ||
| }, | ||
| }, | ||
| { | ||
| Status: clusterv1.MachineStatus{ | ||
| Addresses: clusterv1.MachineAddresses{ | ||
| { | ||
| Address: thirdAddr, | ||
| }, | ||
| }, | ||
| }, | ||
| }, | ||
| } | ||
|
|
||
| opts := clusteragent.Options{ | ||
| IgnoreNodeIPs: sets.NewString(ignoreAddr), | ||
| Port: port, | ||
| } | ||
|
|
||
| // NOTE(Hue): Repeat the test to make sure the IP is not picked by chance (reduce flakiness). | ||
| for i := 0; i < 30; i++ { | ||
| c, err := clusteragent.NewClient(machines, opts) | ||
|
|
||
| g.Expect(err).ToNot(HaveOccurred()) | ||
|
|
||
| // Check if the endpoint is one of the expected ones and not the ignored one. | ||
| g.Expect([]string{fmt.Sprintf("https://%s:%s", firstAddr, port), fmt.Sprintf("https://%s:%s", secondAddr, port), fmt.Sprintf("https://%s:%s", thirdAddr, port)}).To(ContainElement(c.Endpoint())) | ||
| g.Expect(c.Endpoint()).ToNot(Equal(fmt.Sprintf("https://%s:%s", ignoreAddr, port))) | ||
| } | ||
|
|
||
| }) | ||
| } | ||
|
|
||
| func TestDo(t *testing.T) { | ||
| g := NewWithT(t) | ||
|
|
||
| path := "/random/path" | ||
| method := http.MethodPost | ||
| resp := map[string]string{ | ||
| "key": "value", | ||
| } | ||
| servM := NewServerMock(method, path, resp) | ||
| defer servM.ts.Close() | ||
|
|
||
| ip, port, err := net.SplitHostPort(strings.TrimPrefix(servM.ts.URL, "https://")) | ||
| g.Expect(err).ToNot(HaveOccurred()) | ||
| c, err := clusteragent.NewClient([]clusterv1.Machine{ | ||
| { | ||
| Status: clusterv1.MachineStatus{ | ||
| Addresses: clusterv1.MachineAddresses{ | ||
| { | ||
| Address: ip, | ||
| }, | ||
| }, | ||
| }, | ||
| }, | ||
| }, clusteragent.Options{Port: port, InsecureSkipVerify: true}) | ||
|
|
||
| g.Expect(err).ToNot(HaveOccurred()) | ||
|
|
||
| response := make(map[string]string) | ||
| req := map[string]string{"req": "value"} | ||
| path = strings.TrimPrefix(path, "/") | ||
| g.Expect(c.Do(context.Background(), method, path, req, &response)).To(Succeed()) | ||
|
|
||
| g.Expect(response).To(Equal(resp)) | ||
| } | ||
|
|
||
| type serverMock struct { | ||
| method string | ||
| path string | ||
| response any | ||
| request map[string]any | ||
| ts *httptest.Server | ||
| } | ||
|
|
||
| // NewServerMock creates a test server that responds with the given response when called with the given method and path. | ||
| // Make sure to close the server after the test is done. | ||
| // Server will try to decode the request body into a map[string]any. | ||
| func NewServerMock(method string, path string, response any) *serverMock { | ||
| req := make(map[string]any) | ||
| ts := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { | ||
| if r.URL.Path != path { | ||
| http.NotFound(w, r) | ||
| return | ||
| } | ||
| if r.Method != method { | ||
| w.WriteHeader(http.StatusMethodNotAllowed) | ||
| return | ||
| } | ||
|
|
||
| if err := json.NewDecoder(r.Body).Decode(&req); err != nil { | ||
| w.WriteHeader(http.StatusBadRequest) | ||
| return | ||
| } | ||
|
|
||
| if response != nil { | ||
| if err := json.NewEncoder(w).Encode(response); err != nil { | ||
| w.WriteHeader(http.StatusInternalServerError) | ||
| return | ||
| } | ||
| } | ||
| w.WriteHeader(http.StatusOK) | ||
| })) | ||
|
|
||
| return &serverMock{ | ||
| method: method, | ||
| path: path, | ||
| response: response, | ||
| request: req, | ||
| ts: ts, | ||
| } | ||
| } |
Oops, something went wrong.