6234711388

Committed 19 Sep 2023 11:05AM UTC coverage: 27.165%. Remained the same

Build # 6234711388

Build Type

Pull #179

github

Committed by

kemingy

Commit Message

fix fmt

Signed-off-by: Keming <kemingyang@tensorchord.ai>

Pull Request Pull Request #179: fix: avoid json unmarshal error in agent client

Run Details

3 of 3 new or added lines in 1 file covered. (100.0%)

982 of 3615 relevant lines covered (27.16%)

1.66 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/agent/pkg/server/handler_inference_proxy.go

package server

import (
        "errors"
        "fmt"
        "net"
        "net/http"
        "net/http/httputil"
        "strconv"
        "strings"
        "time"

        "github.com/gin-gonic/gin"
        "github.com/prometheus/client_golang/prometheus"

        "github.com/tensorchord/openmodelz/agent/errdefs"
)

// @Summary     Inference.
// @Description Inference proxy.
// @Tags        inference-proxy
// @Accept      json
// @Produce     json
// @Param       name path string true "inference id"
// @Router      /inference/{name} [post]
// @Router      /inference/{name} [get]
// @Router      /inference/{name} [put]
// @Router      /inference/{name} [delete]
// @Success     200
// @Failure     303
// @Failure     400
// @Failure     404
// @Failure     500
func (s *Server) handleInferenceProxy(c *gin.Context) error {
        namespacedName := c.Param("name")
        if namespacedName == "" {
                return NewError(
                        http.StatusBadRequest, errors.New("name is required"), "inference-proxy")
        }

        namespace, name, err := getNamespaceAndName(namespacedName)
        if err != nil {
                return NewError(
                        http.StatusBadRequest, err, "inference-proxy")
        }

        // Update metrics.
        s.metricsOptions.GatewayInferenceInvocationStarted.
                WithLabelValues(namespacedName).Inc()
        s.metricsOptions.GatewayInferenceInvocationInflight.
                WithLabelValues(namespacedName).Inc()
        start := time.Now()
        label := prometheus.Labels{"inference_name": namespacedName, "code": strconv.Itoa(http.StatusProcessing)}
        defer func() {
                s.metricsOptions.GatewayInferenceInvocationInflight.
                        WithLabelValues(namespacedName).Dec()
                s.metricsOptions.GatewayInferencesHistogram.With(label).
                        Observe(time.Since(start).Seconds())
                s.metricsOptions.GatewayInferenceInvocation.With(label).Inc()
        }()

        res := s.scaler.Scale(c.Request.Context(), namespace, name)
        if !res.Found {
                label["code"] = strconv.Itoa(http.StatusNotFound)
                return NewError(
                        http.StatusNotFound, errors.New("inference not found"), "inference-proxy")
        } else if res.Error != nil {
                label["code"] = strconv.Itoa(http.StatusInternalServerError)
                return NewError(
                        http.StatusInternalServerError, res.Error, "inference-proxy")
        }

        if res.Available {
                statusCode, err := s.forward(c, namespace, name)
                if err != nil {
                        label["code"] = strconv.Itoa(statusCode)
                        return NewError(statusCode, err, "inference-proxy")
                }
                label["code"] = strconv.Itoa(statusCode)
                return nil
        } else {
                // The inference is still being created.
                label["code"] = strconv.Itoa(http.StatusSeeOther)
                return NewError(http.StatusSeeOther,
                        fmt.Errorf("inference %s is not available", name), "inference-proxy")
        }
}

func (s *Server) forward(c *gin.Context, namespace, name string) (int, error) {
        backendURL, err := s.endpointResolver.Resolve(namespace, name)
        if err != nil {
                return 0, errdefs.InvalidParameter(err)
        }
        defer s.endpointResolver.Close(backendURL)

        proxyServer := httputil.ReverseProxy{}
        proxyServer.Transport = &http.Transport{
                Proxy: http.ProxyFromEnvironment,
                DialContext: (&net.Dialer{
                        Timeout:   s.config.Server.ReadTimeout,
                        KeepAlive: s.config.Server.ReadTimeout,
                        DualStack: true,
                }).DialContext,
        }
        proxyServer.Director = func(req *http.Request) {
                targetQuery := backendURL.RawQuery
                req.URL.Scheme = backendURL.Scheme
                req.URL.Host = backendURL.Host
                if targetQuery == "" || req.URL.RawQuery == "" {
                        req.URL.RawQuery = targetQuery + req.URL.RawQuery
                } else {
                        req.URL.RawQuery = targetQuery + "&" + req.URL.RawQuery
                }
                req.URL.Path = c.Param("proxyPath")
                if req.URL.Path == "" {
                        req.URL.Path = "/"
                }

                s.logger.WithField("url", backendURL.String()).
                        WithField("path", req.URL.Path).
                        WithField("header", req.Header).
                        WithField("raw-query", req.URL.RawQuery).Debug("reverse proxy")
        }

        var statusCode int
        proxyServer.ModifyResponse = func(resp *http.Response) error {
                statusCode = resp.StatusCode
                return nil
        }

        proxyServer.ServeHTTP(c.Writer, c.Request)
        return statusCode, nil
}

func getNamespaceAndName(name string) (string, string, error) {
        if !strings.Contains(name, ".") {
                return "", "", fmt.Errorf("name is not namespaced")
        }
        namespace := name[strings.LastIndexAny(name, ".")+1:]
        infName := strings.TrimSuffix(name, "."+namespace)

        if namespace == "" {
                return "", "", fmt.Errorf("namespace is empty")
        }

        if infName == "" {
                return "", "", fmt.Errorf("inference name is empty")
        }
        return namespace, infName, nil
}

1	package server
2
3	import (
4	"errors"
5	"fmt"
6	"net"
7	"net/http"
8	"net/http/httputil"
9	"strconv"
10	"strings"
11	"time"
12
13	"github.com/gin-gonic/gin"
14	"github.com/prometheus/client_golang/prometheus"
15
16	"github.com/tensorchord/openmodelz/agent/errdefs"
17	)
18
19	// @Summary Inference.
20	// @Description Inference proxy.
21	// @Tags inference-proxy
22	// @Accept json
23	// @Produce json
24	// @Param name path string true "inference id"
25	// @Router /inference/{name} [post]
26	// @Router /inference/{name} [get]
27	// @Router /inference/{name} [put]
28	// @Router /inference/{name} [delete]
29	// @Success 200
30	// @Failure 303
31	// @Failure 400
32	// @Failure 404
33	// @Failure 500
34	func (s Server) handleInferenceProxy(c gin.Context) error {	×
35	namespacedName := c.Param("name")	×
36	if namespacedName == "" {	×
37	return NewError(	×
38	http.StatusBadRequest, errors.New("name is required"), "inference-proxy")	×
39	}	×
40
41	namespace, name, err := getNamespaceAndName(namespacedName)	×
42	if err != nil {	×
43	return NewError(	×
44	http.StatusBadRequest, err, "inference-proxy")	×
45	}	×
46
47	// Update metrics.
48	s.metricsOptions.GatewayInferenceInvocationStarted.	×
49	WithLabelValues(namespacedName).Inc()	×
50	s.metricsOptions.GatewayInferenceInvocationInflight.	×
51	WithLabelValues(namespacedName).Inc()	×
52	start := time.Now()	×
53	label := prometheus.Labels{"inference_name": namespacedName, "code": strconv.Itoa(http.StatusProcessing)}	×
54	defer func() {	×
55	s.metricsOptions.GatewayInferenceInvocationInflight.	×
56	WithLabelValues(namespacedName).Dec()	×
57	s.metricsOptions.GatewayInferencesHistogram.With(label).	×
58	Observe(time.Since(start).Seconds())	×
59	s.metricsOptions.GatewayInferenceInvocation.With(label).Inc()	×
60	}()	×
61
62	res := s.scaler.Scale(c.Request.Context(), namespace, name)	×
63	if !res.Found {	×
64	label["code"] = strconv.Itoa(http.StatusNotFound)	×
65	return NewError(	×
66	http.StatusNotFound, errors.New("inference not found"), "inference-proxy")	×
67	} else if res.Error != nil {	×
68	label["code"] = strconv.Itoa(http.StatusInternalServerError)	×
69	return NewError(	×
70	http.StatusInternalServerError, res.Error, "inference-proxy")	×
71	}	×
72
73	if res.Available {	×
74	statusCode, err := s.forward(c, namespace, name)	×
75	if err != nil {	×
76	label["code"] = strconv.Itoa(statusCode)	×
77	return NewError(statusCode, err, "inference-proxy")	×
78	}	×
79	label["code"] = strconv.Itoa(statusCode)	×
80	return nil	×
81	} else {	×
82	// The inference is still being created.	×
83	label["code"] = strconv.Itoa(http.StatusSeeOther)	×
84	return NewError(http.StatusSeeOther,	×
85	fmt.Errorf("inference %s is not available", name), "inference-proxy")	×
86	}	×
87	}
88
89	func (s Server) forward(c gin.Context, namespace, name string) (int, error) {	×
90	backendURL, err := s.endpointResolver.Resolve(namespace, name)	×
91	if err != nil {	×
92	return 0, errdefs.InvalidParameter(err)	×
93	}	×
94	defer s.endpointResolver.Close(backendURL)	×
95		×
96	proxyServer := httputil.ReverseProxy{}	×
97	proxyServer.Transport = &http.Transport{	×
98	Proxy: http.ProxyFromEnvironment,	×
99	DialContext: (&net.Dialer{	×
100	Timeout: s.config.Server.ReadTimeout,	×
101	KeepAlive: s.config.Server.ReadTimeout,	×
102	DualStack: true,	×
103	}).DialContext,	×
104	}	×
105	proxyServer.Director = func(req *http.Request) {	×
106	targetQuery := backendURL.RawQuery	×
107	req.URL.Scheme = backendURL.Scheme	×
108	req.URL.Host = backendURL.Host	×
109	if targetQuery == "" \|\| req.URL.RawQuery == "" {	×
110	req.URL.RawQuery = targetQuery + req.URL.RawQuery	×
111	} else {	×
112	req.URL.RawQuery = targetQuery + "&" + req.URL.RawQuery	×
113	}	×
114	req.URL.Path = c.Param("proxyPath")	×
115	if req.URL.Path == "" {	×
116	req.URL.Path = "/"	×
117	}	×
118
119	s.logger.WithField("url", backendURL.String()).	×
120	WithField("path", req.URL.Path).	×
121	WithField("header", req.Header).	×
122	WithField("raw-query", req.URL.RawQuery).Debug("reverse proxy")	×
123	}
124
125	var statusCode int	×
126	proxyServer.ModifyResponse = func(resp *http.Response) error {	×
127	statusCode = resp.StatusCode	×
128	return nil	×
129	}	×
130
131	proxyServer.ServeHTTP(c.Writer, c.Request)	×
132	return statusCode, nil	×
133	}
134
135	func getNamespaceAndName(name string) (string, string, error) {	×
136	if !strings.Contains(name, ".") {	×
137	return "", "", fmt.Errorf("name is not namespaced")	×
138	}	×
139	namespace := name[strings.LastIndexAny(name, ".")+1:]	×
140	infName := strings.TrimSuffix(name, "."+namespace)	×
141		×
142	if namespace == "" {	×
143	return "", "", fmt.Errorf("namespace is empty")	×
144	}	×
145
146	if infName == "" {	×
147	return "", "", fmt.Errorf("inference name is empty")	×
148	}	×
149	return namespace, infName, nil	×
150	}

tensorchord / openmodelz / 6234711388

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous