Skip to content

Commit 732773e

Browse files
miladhzzzzmiladhzzzz
committed
Implement SDK HTTP gateway transport
1 parent 64b5bae commit 732773e

2 files changed

Lines changed: 166 additions & 6 deletions

File tree

‎sdk/client/client.go‎

Lines changed: 127 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -2,16 +2,22 @@
22
package client
33

44
import (
5+
"bytes"
56
"context"
67
"fmt"
8+
"io"
79
"net/http"
10+
"net/url"
11+
"strings"
812
"time"
913

1014
controlv1 "github.com/persys-dev/persys-cloud/pkg/scheduler/controlv1"
1115
"github.com/persys-dev/persys-cloud/sdk/options"
1216
"google.golang.org/grpc"
1317
"google.golang.org/grpc/credentials"
1418
"google.golang.org/grpc/credentials/insecure"
19+
"google.golang.org/protobuf/encoding/protojson"
20+
"google.golang.org/protobuf/proto"
1521
)
1622

1723
// Options is re-exported for callers that import only the client package.
@@ -62,20 +68,83 @@ func New(opts *options.Options) (*Client, error) {
6268
}
6369
dialOpts = append(dialOpts, grpc.WithTransportCredentials(credentials.NewTLS(tlsCfg)))
6470
}
65-
conn, err := grpc.NewClient(opts.GRPCEndpoint, dialOpts...)
71+
conn, err := grpc.DialContext(context.Background(), opts.GRPCEndpoint, dialOpts...)
6672
if err != nil {
6773
return nil, fmt.Errorf("create grpc client: %w", err)
6874
}
6975
c.conn = conn
7076
c.grpc = controlv1.NewAgentControlClient(conn)
7177
case options.TransportHTTP:
72-
return nil, fmt.Errorf("http transport is not implemented yet")
78+
if strings.TrimSpace(opts.APIEndpoint) == "" {
79+
return nil, fmt.Errorf("api endpoint is required for http transport")
80+
}
7381
default:
7482
return nil, fmt.Errorf("unsupported transport %q", opts.Transport)
7583
}
7684
return c, nil
7785
}
7886

87+
func (c *Client) httpURL(path string, query map[string]string) (string, error) {
88+
if c == nil || c.opts == nil {
89+
return "", fmt.Errorf("client is not configured")
90+
}
91+
base, err := url.Parse(strings.TrimRight(c.opts.APIEndpoint, "/"))
92+
if err != nil {
93+
return "", fmt.Errorf("parse api endpoint: %w", err)
94+
}
95+
base.Path = strings.TrimRight(base.Path, "/") + path
96+
values := base.Query()
97+
for k, v := range query {
98+
if strings.TrimSpace(v) != "" {
99+
values.Set(k, v)
100+
}
101+
}
102+
base.RawQuery = values.Encode()
103+
return base.String(), nil
104+
}
105+
106+
func (c *Client) doProtoHTTP(ctx context.Context, method, path string, in proto.Message, out proto.Message, query map[string]string) error {
107+
endpoint, err := c.httpURL(path, query)
108+
if err != nil {
109+
return err
110+
}
111+
var body io.Reader
112+
if in != nil {
113+
data, err := protojson.MarshalOptions{UseProtoNames: true}.Marshal(in)
114+
if err != nil {
115+
return fmt.Errorf("encode request: %w", err)
116+
}
117+
body = bytes.NewReader(data)
118+
}
119+
req, err := http.NewRequestWithContext(ctx, method, endpoint, body)
120+
if err != nil {
121+
return fmt.Errorf("create http request: %w", err)
122+
}
123+
if in != nil {
124+
req.Header.Set("Content-Type", "application/json")
125+
}
126+
req.Header.Set("Accept", "application/json")
127+
resp, err := c.http.Do(req)
128+
if err != nil {
129+
return fmt.Errorf("send http request: %w", err)
130+
}
131+
defer resp.Body.Close()
132+
data, err := io.ReadAll(resp.Body)
133+
if err != nil {
134+
return fmt.Errorf("read http response: %w", err)
135+
}
136+
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
137+
return fmt.Errorf("http %s %s failed with status %d: %s", method, path, resp.StatusCode, strings.TrimSpace(string(data)))
138+
}
139+
if out == nil || len(strings.TrimSpace(string(data))) == 0 {
140+
return nil
141+
}
142+
if err := (protojson.UnmarshalOptions{DiscardUnknown: true}).Unmarshal(data, out); err != nil {
143+
return fmt.Errorf("decode response: %w", err)
144+
}
145+
return nil
146+
}
147+
79148
// Close releases client resources.
80149
func (c *Client) Close() error {
81150
if c != nil && c.conn != nil {
@@ -93,6 +162,13 @@ func (c *Client) requireGRPC() (controlv1.AgentControlClient, error) {
93162

94163
// ApplyWorkload applies or updates a workload.
95164
func (c *Client) ApplyWorkload(ctx context.Context, req *controlv1.ApplyWorkloadRequest) (*controlv1.ApplyWorkloadResponse, error) {
165+
if c != nil && c.opts != nil && c.opts.Transport == options.TransportHTTP {
166+
resp := &controlv1.ApplyWorkloadResponse{}
167+
if err := c.doProtoHTTP(ctx, http.MethodPost, "/workloads/schedule", req, resp, nil); err != nil {
168+
return nil, fmt.Errorf("apply workload: %w", err)
169+
}
170+
return resp, nil
171+
}
96172
gc, err := c.requireGRPC()
97173
if err != nil {
98174
return nil, err
@@ -106,6 +182,13 @@ func (c *Client) ApplyWorkload(ctx context.Context, req *controlv1.ApplyWorkload
106182

107183
// DeleteWorkload deletes a workload.
108184
func (c *Client) DeleteWorkload(ctx context.Context, req *controlv1.DeleteWorkloadRequest) (*controlv1.DeleteWorkloadResponse, error) {
185+
if c != nil && c.opts != nil && c.opts.Transport == options.TransportHTTP {
186+
resp := &controlv1.DeleteWorkloadResponse{}
187+
if err := c.doProtoHTTP(ctx, http.MethodDelete, "/workloads/"+url.PathEscape(req.GetWorkloadId()), nil, resp, nil); err != nil {
188+
return nil, fmt.Errorf("delete workload: %w", err)
189+
}
190+
return resp, nil
191+
}
109192
gc, err := c.requireGRPC()
110193
if err != nil {
111194
return nil, err
@@ -119,6 +202,13 @@ func (c *Client) DeleteWorkload(ctx context.Context, req *controlv1.DeleteWorklo
119202

120203
// RetryWorkload retries a workload.
121204
func (c *Client) RetryWorkload(ctx context.Context, req *controlv1.RetryWorkloadRequest) (*controlv1.RetryWorkloadResponse, error) {
205+
if c != nil && c.opts != nil && c.opts.Transport == options.TransportHTTP {
206+
resp := &controlv1.RetryWorkloadResponse{}
207+
if err := c.doProtoHTTP(ctx, http.MethodPost, "/workloads/"+url.PathEscape(req.GetWorkloadId())+"/retry", nil, resp, nil); err != nil {
208+
return nil, fmt.Errorf("retry workload: %w", err)
209+
}
210+
return resp, nil
211+
}
122212
gc, err := c.requireGRPC()
123213
if err != nil {
124214
return nil, err
@@ -132,6 +222,13 @@ func (c *Client) RetryWorkload(ctx context.Context, req *controlv1.RetryWorkload
132222

133223
// ListNodes lists scheduler nodes.
134224
func (c *Client) ListNodes(ctx context.Context, req *controlv1.ListNodesRequest) (*controlv1.ListNodesResponse, error) {
225+
if c != nil && c.opts != nil && c.opts.Transport == options.TransportHTTP {
226+
resp := &controlv1.ListNodesResponse{}
227+
if err := c.doProtoHTTP(ctx, http.MethodGet, "/nodes", nil, resp, map[string]string{"status": req.GetStatus()}); err != nil {
228+
return nil, fmt.Errorf("list nodes: %w", err)
229+
}
230+
return resp, nil
231+
}
135232
gc, err := c.requireGRPC()
136233
if err != nil {
137234
return nil, err
@@ -145,6 +242,13 @@ func (c *Client) ListNodes(ctx context.Context, req *controlv1.ListNodesRequest)
145242

146243
// GetNode gets a node.
147244
func (c *Client) GetNode(ctx context.Context, req *controlv1.GetNodeRequest) (*controlv1.GetNodeResponse, error) {
245+
if c != nil && c.opts != nil && c.opts.Transport == options.TransportHTTP {
246+
resp := &controlv1.GetNodeResponse{}
247+
if err := c.doProtoHTTP(ctx, http.MethodGet, "/nodes/"+url.PathEscape(req.GetNodeId()), nil, resp, nil); err != nil {
248+
return nil, fmt.Errorf("get node: %w", err)
249+
}
250+
return resp, nil
251+
}
148252
gc, err := c.requireGRPC()
149253
if err != nil {
150254
return nil, err
@@ -158,6 +262,13 @@ func (c *Client) GetNode(ctx context.Context, req *controlv1.GetNodeRequest) (*c
158262

159263
// ListWorkloads lists workloads.
160264
func (c *Client) ListWorkloads(ctx context.Context, req *controlv1.ListWorkloadsRequest) (*controlv1.ListWorkloadsResponse, error) {
265+
if c != nil && c.opts != nil && c.opts.Transport == options.TransportHTTP {
266+
resp := &controlv1.ListWorkloadsResponse{}
267+
if err := c.doProtoHTTP(ctx, http.MethodGet, "/workloads", nil, resp, map[string]string{"status": req.GetStatus()}); err != nil {
268+
return nil, fmt.Errorf("list workloads: %w", err)
269+
}
270+
return resp, nil
271+
}
161272
gc, err := c.requireGRPC()
162273
if err != nil {
163274
return nil, err
@@ -171,6 +282,13 @@ func (c *Client) ListWorkloads(ctx context.Context, req *controlv1.ListWorkloads
171282

172283
// GetWorkload gets a workload.
173284
func (c *Client) GetWorkload(ctx context.Context, req *controlv1.GetWorkloadRequest) (*controlv1.GetWorkloadResponse, error) {
285+
if c != nil && c.opts != nil && c.opts.Transport == options.TransportHTTP {
286+
resp := &controlv1.GetWorkloadResponse{}
287+
if err := c.doProtoHTTP(ctx, http.MethodGet, "/workloads/"+url.PathEscape(req.GetWorkloadId()), nil, resp, nil); err != nil {
288+
return nil, fmt.Errorf("get workload: %w", err)
289+
}
290+
return resp, nil
291+
}
174292
gc, err := c.requireGRPC()
175293
if err != nil {
176294
return nil, err
@@ -184,6 +302,13 @@ func (c *Client) GetWorkload(ctx context.Context, req *controlv1.GetWorkloadRequ
184302

185303
// GetClusterSummary gets cluster status.
186304
func (c *Client) GetClusterSummary(ctx context.Context, req *controlv1.GetClusterSummaryRequest) (*controlv1.GetClusterSummaryResponse, error) {
305+
if c != nil && c.opts != nil && c.opts.Transport == options.TransportHTTP {
306+
resp := &controlv1.GetClusterSummaryResponse{}
307+
if err := c.doProtoHTTP(ctx, http.MethodGet, "/cluster/metrics", nil, resp, nil); err != nil {
308+
return nil, fmt.Errorf("get cluster summary: %w", err)
309+
}
310+
return resp, nil
311+
}
187312
gc, err := c.requireGRPC()
188313
if err != nil {
189314
return nil, err

‎sdk/client/client_test.go‎

Lines changed: 39 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,14 @@
11
package client
22

33
import (
4+
"context"
5+
"net/http"
6+
"net/http/httptest"
47
"testing"
58

9+
controlv1 "github.com/persys-dev/persys-cloud/pkg/scheduler/controlv1"
610
"github.com/persys-dev/persys-cloud/sdk/options"
11+
"google.golang.org/protobuf/encoding/protojson"
712
)
813

914
func TestNewGRPCInsecure(t *testing.T) {
@@ -26,10 +31,40 @@ func TestNewRejectsUnknownTransport(t *testing.T) {
2631
}
2732
}
2833

29-
func TestNewRejectsHTTPTransportUntilImplemented(t *testing.T) {
30-
_, err := New(&options.Options{Transport: options.TransportHTTP, APIEndpoint: "http://localhost:8080"})
31-
if err == nil {
32-
t.Fatal("New() expected error for HTTP transport")
34+
func TestNewHTTPTransport(t *testing.T) {
35+
c, err := New(&options.Options{Transport: options.TransportHTTP, APIEndpoint: "http://localhost:8080"})
36+
if err != nil {
37+
t.Fatalf("New() error = %v", err)
38+
}
39+
if c == nil {
40+
t.Fatal("New() returned nil client")
41+
}
42+
}
43+
44+
func TestHTTPApplyWorkloadUsesGatewayEndpoint(t *testing.T) {
45+
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
46+
if r.Method != http.MethodPost || r.URL.Path != "/workloads/schedule" {
47+
t.Fatalf("unexpected request %s %s", r.Method, r.URL.Path)
48+
}
49+
w.Header().Set("Content-Type", "application/json")
50+
data, err := protojson.MarshalOptions{UseProtoNames: true}.Marshal(&controlv1.ApplyWorkloadResponse{Success: true})
51+
if err != nil {
52+
t.Fatalf("marshal response: %v", err)
53+
}
54+
_, _ = w.Write(data)
55+
}))
56+
defer server.Close()
57+
58+
c, err := New(&options.Options{Transport: options.TransportHTTP, APIEndpoint: server.URL})
59+
if err != nil {
60+
t.Fatalf("New() error = %v", err)
61+
}
62+
resp, err := c.ApplyWorkload(context.Background(), &controlv1.ApplyWorkloadRequest{WorkloadId: "workload-1"})
63+
if err != nil {
64+
t.Fatalf("ApplyWorkload() error = %v", err)
65+
}
66+
if !resp.GetSuccess() {
67+
t.Fatal("ApplyWorkload() returned unsuccessful response")
3368
}
3469
}
3570

0 commit comments

Comments
 (0)