- 
                Notifications
    You must be signed in to change notification settings 
- Fork 256
[YUNIKORN-2249] Add compression option to getQueueApplication API #757
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
base: master
Are you sure you want to change the base?
Changes from 19 commits
9ef7968
              85dd3de
              b5e2651
              e7ea5bc
              7238626
              ce40af6
              995cb97
              4cb1d9c
              40b9649
              4f57568
              4e7228a
              2205902
              b811993
              8a6f9cf
              53c8827
              1c4a0e1
              41c0570
              ca18e79
              0a6c931
              3f806ba
              File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change | 
|---|---|---|
| @@ -0,0 +1,125 @@ | ||
| /* | ||
| Licensed to the Apache Software Foundation (ASF) under one | ||
| or more contributor license agreements. See the NOTICE file | ||
| distributed with this work for additional information | ||
| regarding copyright ownership. The ASF licenses this file | ||
| to you under the Apache License, Version 2.0 (the | ||
| "License"); you may not use this file except in compliance | ||
| with the License. You may obtain a copy of the License at | ||
| http://www.apache.org/licenses/LICENSE-2.0 | ||
| Unless required by applicable law or agreed to in writing, software | ||
| distributed under the License is distributed on an "AS IS" BASIS, | ||
| WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| See the License for the specific language governing permissions and | ||
| limitations under the License. | ||
| */ | ||
|  | ||
| package webservice | ||
|  | ||
| import ( | ||
| "compress/gzip" | ||
| "encoding/json" | ||
| "io" | ||
| "net" | ||
| "net/http" | ||
| "net/url" | ||
| "testing" | ||
| "time" | ||
|  | ||
| "github.com/julienschmidt/httprouter" | ||
| "go.uber.org/zap" | ||
| "gotest.tools/v3/assert" | ||
|  | ||
| "github.com/apache/yunikorn-core/pkg/common" | ||
| "github.com/apache/yunikorn-core/pkg/log" | ||
| "github.com/apache/yunikorn-core/pkg/scheduler" | ||
| ) | ||
|  | ||
| func TestCompressionWithDummyRoute(t *testing.T) { | ||
|         
                  targetoee marked this conversation as resolved.
              Outdated
          
            Show resolved
            Hide resolved | ||
| m := NewWebApp(scheduler.NewScheduler().GetClusterContext(), nil) | ||
| // dummy route and corresponding handler | ||
| testRoute := route{ | ||
| "testHelloWord", | ||
| "GET", | ||
| "/ws/v1/helloWorld", | ||
| getHelloWorld, | ||
| } | ||
| router := httprouter.New() | ||
| testHandler := gzipHandler(loggingHandler(testRoute.HandlerFunc, testRoute.Name)) | ||
| router.Handler(testRoute.Method, testRoute.Pattern, testHandler) | ||
|  | ||
| // start simulation server | ||
| m.httpServer = &http.Server{Addr: ":9080", Handler: router, ReadHeaderTimeout: 5 * time.Second} | ||
| go func() { | ||
| There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Could we reuse the code of  func newRouter(routes []route) *httprouter.Router {
	router := httprouter.New()
	for _, webRoute := range routes {
		handler := gzipHandler(loggingHandler(webRoute.HandlerFunc, webRoute.Name))
		router.Handler(webRoute.Method, webRoute.Pattern, handler)
	}
	return router
}
func (m *WebService) StartWebApp() {
	m.startWebApp(webRoutes)
}
func (m *WebService) startWebApp(routes []route) {
	m.httpServer = &http.Server{Addr: ":9080", Handler: newRouter(routes)}
	log.Log(log.REST).Info("web-app started", zap.Int("port", 9080))
	go func() {
		httpError := m.httpServer.ListenAndServe()
		if httpError != nil && !errors.Is(httpError, http.ErrServerClosed) {
			log.Log(log.REST).Error("HTTP serving error",
				zap.Error(httpError))
		}
	}()
}and then we call  	m := NewWebApp(scheduler.NewScheduler().GetClusterContext(), nil)
	// start simulation server
	m.startWebApp([]route{route{
		"testHelloWord",
		"GET",
		"/ws/v1/helloWorld",
		getHelloWorld,
	}}) | ||
| httpError := m.httpServer.ListenAndServe() | ||
| if httpError != nil { | ||
| log.Log(log.REST).Error("HTTP serving error", | ||
| zap.Error(httpError)) | ||
| } | ||
| }() | ||
| defer func() { | ||
| err := m.StopWebApp() | ||
| assert.NilError(t, err, "Error when closing webapp service.") | ||
| }() | ||
|  | ||
| err := common.WaitFor(500*time.Millisecond, 5*time.Second, func() bool { | ||
| conn, connErr := net.DialTimeout("tcp", net.JoinHostPort("127.0.0.1", "9080"), time.Second) | ||
| if connErr == nil { | ||
| conn.Close() | ||
| } | ||
| return connErr == nil | ||
| }) | ||
| assert.NilError(t, err, "Web app failed to start in 2 seconds.") | ||
|         
                  targetoee marked this conversation as resolved.
              Outdated
          
            Show resolved
            Hide resolved | ||
|  | ||
| u := &url.URL{ | ||
| Host: "localhost:9080", | ||
| Scheme: "http", | ||
| Path: "/ws/v1/helloWorld", | ||
| } | ||
|  | ||
| // request without gzip compression | ||
| var buf io.ReadWriter | ||
| req, err := http.NewRequest("GET", u.String(), buf) | ||
| assert.NilError(t, err, "Create new http request failed.") | ||
| req.Header.Set("Accept", "application/json") | ||
|  | ||
| // prevent http.DefaultClient from automatically adding gzip header | ||
| There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. We can merge those code into a helper function. For example: 	check := func(compression bool) {
		req, err := http.NewRequest("GET", u.String(), nil)
		assert.NilError(t, err, "Create new http request failed.")
		req.Header.Set("Accept", "application/json")
		if compression {
			req.Header.Set("Accept-Encoding", "gzip")
		} else {
			req.Header.Set("Accept-Encoding", "deflate")
		}
		resp, err := http.DefaultClient.Do(req)
		assert.NilError(t, err, "Http request failed.")
		defer resp.Body.Close()
		var reader io.Reader
		if compression {
			gzipReader, err := gzip.NewReader(resp.Body)
			assert.NilError(t, err, "Failed to create gzip reader.")
			defer gzipReader.Close()
			reader = gzipReader
		} else {
			reader = resp.Body
		}
		byteArr, err := io.ReadAll(reader)
		assert.NilError(t, err, "Failed to read body.")
		var respMsg map[string]string
		err = json.Unmarshal(byteArr, &respMsg)
		assert.NilError(t, err, unmarshalError)
		assert.Equal(t, len(respMsg), 1)
		assert.Equal(t, respMsg["data"], "hello world")
	}
	check(false)
	check(true) | ||
| req.Header.Set("Accept-Encoding", "deflate") | ||
| resp, err := http.DefaultClient.Do(req) | ||
| assert.NilError(t, err, "Http request failed.") | ||
| defer resp.Body.Close() | ||
| byteArr, err := io.ReadAll(resp.Body) | ||
| assert.NilError(t, err, "Failed to read body.") | ||
| var respMsg map[string]string | ||
| err = json.Unmarshal(byteArr, &respMsg) | ||
| assert.NilError(t, err, unmarshalError) | ||
| assert.Equal(t, respMsg["data"], "hello world") | ||
|  | ||
| // request with gzip compression enabled | ||
| req.Header.Set("Accept-Encoding", "gzip") | ||
| resp2, err := http.DefaultClient.Do(req) | ||
| assert.NilError(t, err, "Http request failed.") | ||
| defer resp2.Body.Close() | ||
| gzipReader, err := gzip.NewReader(resp2.Body) | ||
| assert.NilError(t, err, "Failed to create gzip reader.") | ||
| compressedData, err := io.ReadAll(gzipReader) | ||
| assert.NilError(t, err, "Failed to read body") | ||
| var respMsg2 map[string]string | ||
| err = json.Unmarshal(compressedData, &respMsg2) | ||
| assert.NilError(t, err, unmarshalError) | ||
| assert.Equal(t, respMsg2["data"], "hello world") | ||
| defer gzipReader.Close() | ||
| } | ||
|  | ||
| func getHelloWorld(w http.ResponseWriter, r *http.Request) { | ||
| writeHeaders(w) | ||
| result := map[string]string{ | ||
| "data": "hello world", | ||
| } | ||
|  | ||
| if err := json.NewEncoder(w).Encode(result); err != nil { | ||
| buildJSONErrorResponse(w, err.Error(), http.StatusInternalServerError) | ||
| } | ||
| } | ||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I don't think we should unconditionally GZIP every response, only those that might be large. I'd suggest adding a "Compress" boolean to the route object in routes.go and use that to determine whether to add the gzip handler.
Uh oh!
There was an error while loading. Please reload this page.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
For example, none of the /debug routes should have it (many of those generate binary data which is not compressible, and/or are simple text that wouldn't benefit). It should probably not be on anything that typically results in < 10 KB of data.