diff --git a/apis/core.oam.dev/v1beta1/resourcetracker_types.go b/apis/core.oam.dev/v1beta1/resourcetracker_types.go index e3fb132f1..503f98b39 100644 --- a/apis/core.oam.dev/v1beta1/resourcetracker_types.go +++ b/apis/core.oam.dev/v1beta1/resourcetracker_types.go @@ -100,6 +100,15 @@ func (in *ResourceTrackerSpec) MarshalJSON() ([]byte, error) { cpy.ManagedResources = nil cpy.Compression.Data = data tmp.Alias = (*Alias)(cpy) + case compression.Zstd: + cpy := in.DeepCopy() + data, err := compression.ZstdObjectToString(in.ManagedResources) + if err != nil { + return nil, err + } + cpy.ManagedResources = nil + cpy.Compression.Data = data + tmp.Alias = (*Alias)(cpy) default: return nil, compression.NewUnsupportedCompressionTypeError(string(in.Compression.Type)) } @@ -124,6 +133,12 @@ func (in *ResourceTrackerSpec) UnmarshalJSON(src []byte) error { return err } tmp.Compression.Data = "" + case compression.Zstd: + tmp.ManagedResources = []ManagedResource{} + if err := compression.UnZstdStringToObject(tmp.Compression.Data, &tmp.ManagedResources); err != nil { + return err + } + tmp.Compression.Data = "" default: return compression.NewUnsupportedCompressionTypeError(string(in.Compression.Type)) } diff --git a/apis/core.oam.dev/v1beta1/resourcetracker_types_test.go b/apis/core.oam.dev/v1beta1/resourcetracker_types_test.go index c4c55c716..fb78e77c2 100644 --- a/apis/core.oam.dev/v1beta1/resourcetracker_types_test.go +++ b/apis/core.oam.dev/v1beta1/resourcetracker_types_test.go @@ -19,6 +19,7 @@ package v1beta1 import ( "encoding/json" "fmt" + "io/ioutil" "strings" "testing" "time" @@ -188,37 +189,153 @@ func TestResourceTracker_ManagedResource(t *testing.T) { } func TestResourceTrackerCompression(t *testing.T) { - size := 1000 + count := 20 r := require.New(t) - rt := &ResourceTracker{} - for i := 0; i < size; i++ { - rt.AddManagedResource(&corev1.ConfigMap{ObjectMeta: metav1.ObjectMeta{Name: fmt.Sprintf("cm%d", i)}}, false, false, "") - rt.AddManagedResource(&corev1.Secret{ObjectMeta: metav1.ObjectMeta{Name: fmt.Sprintf("secret%d", i)}}, true, false, "") + + // Load some real CRDs, and other test data to simulate real use-cases. + // The user must have some large resourcetrackers if they use compression, + // so we load some large CRDs. + var data []string + paths := []string{ + "../../../charts/vela-core/crds/core.oam.dev_applicationrevisions.yaml", + "../../../charts/vela-core/crds/core.oam.dev_applications.yaml", + "../../../charts/vela-core/crds/core.oam.dev_definitionrevisions.yaml", + "../../../charts/vela-core/crds/core.oam.dev_healthscopes.yaml", + "../../../charts/vela-core/crds/core.oam.dev_traitdefinitions.yaml", + "../../../charts/vela-core/crds/core.oam.dev_componentdefinitions.yaml", + "../../../charts/vela-core/crds/core.oam.dev_workloaddefinitions.yaml", + "../../../charts/vela-core/crds/standard.oam.dev_rollouts.yaml", + "../../../charts/vela-core/templates/addon/fluxcd.yaml", + "../../../charts/vela-core/templates/kubevela-controller.yaml", + "../../../charts/vela-core/README.md", + "../../../pkg/velaql/providers/query/testdata/machinelearning.seldon.io_seldondeployments.yaml", + "../../../legacy/charts/vela-core-legacy/crds/standard.oam.dev_podspecworkloads.yaml", } - rt.Spec.Compression.Type = compression.Gzip - t0 := time.Now() - bs, err := json.Marshal(rt) - r.NoError(err) - afterElapsed := time.Since(t0).Nanoseconds() - r.Contains(string(bs), `"type":"gzip","data":`) - _rt := &ResourceTracker{} - r.NoError(json.Unmarshal(bs, _rt)) - r.Equal(size*2, len(_rt.Spec.ManagedResources)) - for i, rsc := range _rt.Spec.ManagedResources { - r.Equal(i%2 == 1, rsc.Data == nil) + for _, p := range paths { + b, err := ioutil.ReadFile(p) + r.NoError(err) + data = append(data, string(b)) + } + size := len(data) + + // Gzip + var ( + gzipCompressTime int64 + gzipSize int + gzipBs []byte + ) + for c := 0; c < count; c++ { + var err error + rtGzip := &ResourceTracker{} + for i := 0; i < size; i++ { + rtGzip.AddManagedResource(&corev1.ConfigMap{ObjectMeta: metav1.ObjectMeta{Name: fmt.Sprintf("cm%d", i)}, Data: map[string]string{"1": data[i]}}, false, false, "") + rtGzip.AddManagedResource(&corev1.Secret{ObjectMeta: metav1.ObjectMeta{Name: fmt.Sprintf("secret%d", i)}}, true, false, "") + } + rtGzip.Spec.Compression.Type = compression.Gzip + // Compress + t0 := time.Now() + gzipBs, err = json.Marshal(rtGzip) + elapsed := time.Since(t0).Nanoseconds() + if gzipCompressTime == 0 { + gzipCompressTime = elapsed + } else { + gzipCompressTime = (elapsed + gzipCompressTime) / 2 + } + if gzipSize == 0 { + gzipSize = len(gzipBs) + } else { + gzipSize = (len(gzipBs) + gzipSize) / 2 + } + r.NoError(err) + r.Contains(string(gzipBs), `"type":"gzip","data":`) } - _rt.Spec.Compression.Type = compression.Uncompressed - t0 = time.Now() - _bs, err := json.Marshal(_rt) - beforeElapsed := time.Since(t0) - r.NoError(err) - before, after := len(_bs), len(bs) - r.Less(after, before) - fmt.Printf("Compression Size:\n before: %d\n after: %d\n rate: %.2f%%\n", - before, after, float64(after)*100.0/float64(before)) - fmt.Printf("Compression Time:\n before: %d ns\n after: %d ns\n rate: %.2f%%\n", - beforeElapsed, afterElapsed, float64(afterElapsed)*100.0/float64(beforeElapsed)) + // Zstd + var ( + zstdCompressTime int64 + zstdSize int + zstdBs []byte + ) + for c := 0; c < count; c++ { + var err error + rtZstd := &ResourceTracker{} + for i := 0; i < size; i++ { + rtZstd.AddManagedResource(&corev1.ConfigMap{ObjectMeta: metav1.ObjectMeta{Name: fmt.Sprintf("cm%d", i)}, Data: map[string]string{"1": data[i]}}, false, false, "") + rtZstd.AddManagedResource(&corev1.Secret{ObjectMeta: metav1.ObjectMeta{Name: fmt.Sprintf("secret%d", i)}}, true, false, "") + } + rtZstd.Spec.Compression.Type = compression.Zstd + t0 := time.Now() + zstdBs, err = json.Marshal(rtZstd) + elapsed := time.Since(t0).Nanoseconds() + if zstdCompressTime == 0 { + zstdCompressTime = elapsed + } else { + zstdCompressTime = (elapsed + zstdCompressTime) / 2 + } + if zstdSize == 0 { + zstdSize = len(zstdBs) + } else { + zstdSize = (len(zstdBs) + zstdSize) / 2 + } + r.NoError(err) + r.Contains(string(zstdBs), `"type":"zstd","data":`) + } + + rtUncmp := &ResourceTracker{} + r.NoError(json.Unmarshal(gzipBs, rtUncmp)) + r.Equal(size*2, len(rtUncmp.Spec.ManagedResources)) + for i, rsc := range rtUncmp.Spec.ManagedResources { + r.Equal(i%2 == 1, rsc.Data == nil) + } + r.NoError(json.Unmarshal(zstdBs, rtUncmp)) + r.Equal(size*2, len(rtUncmp.Spec.ManagedResources)) + for i, rsc := range rtUncmp.Spec.ManagedResources { + r.Equal(i%2 == 1, rsc.Data == nil) + } + // No compression + var ( + uncmpTime int64 + uncmpSize int + ) + rtUncmp.Spec.Compression.Type = compression.Uncompressed + for c := 0; c < count; c++ { + t0 := time.Now() + _bs, err := json.Marshal(rtUncmp) + if uncmpTime == 0 { + uncmpTime = time.Since(t0).Nanoseconds() + } else { + uncmpTime = (time.Since(t0).Nanoseconds() + uncmpTime) / 2 + } + if uncmpSize == 0 { + uncmpSize = len(_bs) + } else { + uncmpSize = (len(_bs) + uncmpSize) / 2 + } + r.NoError(err) + before, after := len(_bs), len(zstdBs) + r.Less(after, before) + before, after = len(_bs), len(gzipBs) + r.Less(after, before) + } + + fmt.Printf(`Compressed Size: + uncompressed: %d bytes 100.00%% + gzip: %d bytes %.2f%% + zstd: %d bytes %.2f%% +`, + uncmpSize, + gzipSize, float64(gzipSize)*100.0/float64(uncmpSize), + zstdSize, float64(zstdSize)*100.0/float64(uncmpSize)) + + fmt.Printf(`Marshal Time: + no compression: %d ns 1.00x + gzip: %d ns %.2fx + zstd: %d ns %.2fx +`, + uncmpTime, + gzipCompressTime, float64(gzipCompressTime)/float64(uncmpTime), + zstdCompressTime, float64(zstdCompressTime)/float64(uncmpTime), + ) } func TestResourceTrackerInvalidMarshal(t *testing.T) { diff --git a/charts/vela-core/README.md b/charts/vela-core/README.md index 40a299ff5..f0c86dd80 100644 --- a/charts/vela-core/README.md +++ b/charts/vela-core/README.md @@ -81,20 +81,21 @@ helm install --create-namespace -n vela-system kubevela kubevela/vela-core --wai ### KubeVela controller optimization parameters -| Name | Description | Value | -| ------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------- | ------- | -| `optimize.cachedGvks` | Optimize types of resources to be cached. | `""` | -| `optimize.resourceTrackerListOp` | Optimize ResourceTracker List Op by adding index. | `true` | -| `optimize.controllerReconcileLoopReduction` | Optimize ApplicationController reconcile by reducing the number of loops to reconcile application. | `false` | -| `optimize.markWithProb` | Optimize ResourceTracker GC by only run mark with probability. Side effect: outdated ResourceTracker might not be able to be removed immediately. | `0.1` | -| `optimize.disableComponentRevision` | Optimize componentRevision by disabling the creation and gc | `false` | -| `optimize.disableApplicationRevision` | Optimize ApplicationRevision by disabling the creation and gc. | `false` | -| `optimize.disableWorkflowRecorder` | Optimize workflow recorder by disabling the creation and gc. | `false` | -| `optimize.enableInMemoryWorkflowContext` | Optimize workflow by use in-memory context. | `false` | -| `optimize.disableResourceApplyDoubleCheck` | Optimize workflow by ignoring resource double check after apply. | `false` | -| `optimize.enableResourceTrackerDeleteOnlyTrigger` | Optimize resourcetracker by only trigger reconcile when resourcetracker is deleted. | `true` | -| `featureGates.enableLegacyComponentRevision` | if disabled, only component with rollout trait will create component revisions | `false` | -| `featureGates.gzipResourceTracker` | if enabled, resourceTracker will be compressed before stored | `false` | +| Name | Description | Value | +| ------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | ------- | +| `optimize.cachedGvks` | Optimize types of resources to be cached. | `""` | +| `optimize.resourceTrackerListOp` | Optimize ResourceTracker List Op by adding index. | `true` | +| `optimize.controllerReconcileLoopReduction` | Optimize ApplicationController reconcile by reducing the number of loops to reconcile application. | `false` | +| `optimize.markWithProb` | Optimize ResourceTracker GC by only run mark with probability. Side effect: outdated ResourceTracker might not be able to be removed immediately. | `0.1` | +| `optimize.disableComponentRevision` | Optimize componentRevision by disabling the creation and gc | `false` | +| `optimize.disableApplicationRevision` | Optimize ApplicationRevision by disabling the creation and gc. | `false` | +| `optimize.disableWorkflowRecorder` | Optimize workflow recorder by disabling the creation and gc. | `false` | +| `optimize.enableInMemoryWorkflowContext` | Optimize workflow by use in-memory context. | `false` | +| `optimize.disableResourceApplyDoubleCheck` | Optimize workflow by ignoring resource double check after apply. | `false` | +| `optimize.enableResourceTrackerDeleteOnlyTrigger` | Optimize resourcetracker by only trigger reconcile when resourcetracker is deleted. | `true` | +| `featureGates.enableLegacyComponentRevision` | if disabled, only component with rollout trait will create component revisions | `false` | +| `featureGates.gzipResourceTracker` | if enabled, resourceTracker will be compressed using gzip before being stored | `false` | +| `featureGates.zstdResourceTracker` | if enabled, resourceTracker will be compressed using zstd before being stored. It is much faster and more efficient than gzip. If both gzip and zstd are enabled, zstd will be used. | `false` | ### MultiCluster parameters diff --git a/charts/vela-core/templates/kubevela-controller.yaml b/charts/vela-core/templates/kubevela-controller.yaml index be57db998..f6885000a 100644 --- a/charts/vela-core/templates/kubevela-controller.yaml +++ b/charts/vela-core/templates/kubevela-controller.yaml @@ -219,6 +219,7 @@ spec: - "--feature-gates=AuthenticateApplication={{- .Values.authentication.enabled | toString -}}" - "--feature-gates=LegacyComponentRevision={{- .Values.featureGates.enableLegacyComponentRevision | toString -}}" - "--feature-gates=GzipResourceTracker={{- .Values.featureGates.gzipResourceTracker | toString -}}" + - "--feature-gates=ZstdResourceTracker={{- .Values.featureGates.zstdResourceTracker | toString -}}" {{ if .Values.authentication.enabled }} {{ if .Values.authentication.withUser }} - "--authentication-with-user" diff --git a/charts/vela-core/values.yaml b/charts/vela-core/values.yaml index 9b2913044..6585a8800 100644 --- a/charts/vela-core/values.yaml +++ b/charts/vela-core/values.yaml @@ -110,10 +110,12 @@ optimize: enableResourceTrackerDeleteOnlyTrigger: true ##@param featureGates.enableLegacyComponentRevision if disabled, only component with rollout trait will create component revisions -##@param featureGates.gzipResourceTracker if enabled, resourceTracker will be compressed before stored +##@param featureGates.gzipResourceTracker if enabled, resourceTracker will be compressed using gzip before being stored +##@param featureGates.zstdResourceTracker if enabled, resourceTracker will be compressed using zstd before being stored. It is much faster and more efficient than gzip. If both gzip and zstd are enabled, zstd will be used. featureGates: enableLegacyComponentRevision: false gzipResourceTracker: false + zstdResourceTracker: false ## @section MultiCluster parameters diff --git a/go.mod b/go.mod index 04af35afa..07560c235 100644 --- a/go.mod +++ b/go.mod @@ -48,6 +48,7 @@ require ( github.com/hashicorp/hcl/v2 v2.9.1 github.com/hinshun/vt10x v0.0.0-20180616224451-1954e6464174 github.com/imdario/mergo v0.3.12 + github.com/klauspost/compress v1.15.9 github.com/koding/websocketproxy v0.0.0-20181220232114-7ed82d81a28c github.com/kubevela/prism v1.4.1-0.20220613123457-94f1190f87c2 github.com/kyokomi/emoji v2.2.4+incompatible @@ -64,6 +65,7 @@ require ( github.com/pkg/errors v0.9.1 github.com/prometheus/client_golang v1.11.0 github.com/robfig/cron/v3 v3.0.1 + github.com/rogpeppe/go-internal v1.8.1 github.com/sirupsen/logrus v1.8.1 github.com/spf13/cobra v1.4.0 github.com/spf13/pflag v1.0.5 @@ -129,8 +131,6 @@ require ( github.com/rivo/tview v0.0.0-20220709181631-73bf2902b59a ) -require github.com/rogpeppe/go-internal v1.8.1 - require ( cloud.google.com/go/compute v1.7.0 // indirect github.com/Azure/go-ansiterm v0.0.0-20210617225240-d185dfc1b5a1 // indirect @@ -225,7 +225,6 @@ require ( github.com/josharian/intern v1.0.0 // indirect github.com/json-iterator/go v1.1.12 // indirect github.com/kballard/go-shellquote v0.0.0-20180428030007-95032a82bc51 // indirect - github.com/klauspost/compress v1.15.4 // indirect github.com/kr/pretty v0.3.0 // indirect github.com/kr/pty v1.1.8 // indirect github.com/kr/text v0.2.0 // indirect diff --git a/go.sum b/go.sum index b9d564980..49911f320 100644 --- a/go.sum +++ b/go.sum @@ -1349,8 +1349,9 @@ github.com/klauspost/compress v1.11.13/go.mod h1:aoV0uJVorq1K+umq18yTdKaF57EivdY github.com/klauspost/compress v1.13.4/go.mod h1:8dP1Hq4DHOhN9w426knH3Rhby4rFm6D8eO+e+Dq5Gzg= github.com/klauspost/compress v1.13.5/go.mod h1:/3/Vjq9QcHkK5uEr5lBEmyoZ1iFhe47etQ6QUkpK6sk= github.com/klauspost/compress v1.15.1/go.mod h1:/3/Vjq9QcHkK5uEr5lBEmyoZ1iFhe47etQ6QUkpK6sk= -github.com/klauspost/compress v1.15.4 h1:1kn4/7MepF/CHmYub99/nNX8az0IJjfSOU/jbnTVfqQ= github.com/klauspost/compress v1.15.4/go.mod h1:PhcZ0MbTNciWF3rruxRgKxI5NkcHHrHUDtV4Yw2GlzU= +github.com/klauspost/compress v1.15.9 h1:wKRjX6JRtDdrE9qwa4b/Cip7ACOshUI4smpCQanqjSY= +github.com/klauspost/compress v1.15.9/go.mod h1:PhcZ0MbTNciWF3rruxRgKxI5NkcHHrHUDtV4Yw2GlzU= github.com/klauspost/cpuid v0.0.0-20170728055534-ae7887de9fa5/go.mod h1:Pj4uuM528wm8OyEC2QMXAi2YiTZ96dNQPGgoMS4s3ek= github.com/klauspost/crc32 v0.0.0-20161016154125-cb6bfca970f6/go.mod h1:+ZoRqAPRLkC4NPOvfYeR5KNOrY6TD+/sAC3HXPZgDYg= github.com/klauspost/pgzip v1.0.2-0.20170402124221-0bf5dcad4ada/go.mod h1:Ch1tH69qFZu15pkjo5kYi6mth2Zzwzt50oCQKQE9RUs= diff --git a/makefiles/e2e.mk b/makefiles/e2e.mk index 2850d3dbf..b68977c6d 100644 --- a/makefiles/e2e.mk +++ b/makefiles/e2e.mk @@ -5,7 +5,12 @@ e2e-setup-core-pre-hook: .PHONY: e2e-setup-core-post-hook e2e-setup-core-post-hook: kubectl wait --for=condition=Available deployment/kubevela-vela-core -n vela-system --timeout=180s - helm upgrade --install --namespace vela-system --wait oam-rollout --set image.repository=vela-runtime-rollout-test --set image.tag=$(GIT_COMMIT) ./runtime/rollout/charts + helm upgrade --install \ + --namespace vela-system \ + --wait oam-rollout \ + --set image.repository=vela-runtime-rollout-test \ + --set image.tag=$(GIT_COMMIT) \ + ./runtime/rollout/charts go run ./e2e/addon/mock & sleep 15 bin/vela addon enable ./e2e/addon/mock/testdata/fluxcd @@ -13,11 +18,32 @@ e2e-setup-core-post-hook: .PHONY: e2e-setup-core-wo-auth e2e-setup-core-wo-auth: - helm upgrade --install --create-namespace --namespace vela-system --set image.pullPolicy=IfNotPresent --set image.repository=vela-core-test --set applicationRevisionLimit=5 --set dependCheckWait=10s --set image.tag=$(GIT_COMMIT) --wait kubevela ./charts/vela-core + helm upgrade --install \ + --create-namespace \ + --namespace vela-system \ + --set image.pullPolicy=IfNotPresent \ + --set image.repository=vela-core-test \ + --set applicationRevisionLimit=5 \ + --set dependCheckWait=10s \ + --set image.tag=$(GIT_COMMIT) \ + --wait kubevela ./charts/vela-core .PHONY: e2e-setup-core-w-auth e2e-setup-core-w-auth: - helm upgrade --install --create-namespace --namespace vela-system --set image.pullPolicy=IfNotPresent --set image.repository=vela-core-test --set applicationRevisionLimit=5 --set dependCheckWait=10s --set image.tag=$(GIT_COMMIT) --wait kubevela ./charts/vela-core --set authentication.enabled=true --set authentication.withUser=true --set authentication.groupPattern=* --set featureGates.gzipResourceTracker=true + helm upgrade --install \ + --create-namespace \ + --namespace vela-system \ + --set image.pullPolicy=IfNotPresent \ + --set image.repository=vela-core-test \ + --set applicationRevisionLimit=5 \ + --set dependCheckWait=10s \ + --set image.tag=$(GIT_COMMIT) \ + --wait kubevela \ + ./charts/vela-core \ + --set authentication.enabled=true \ + --set authentication.withUser=true \ + --set authentication.groupPattern=* \ + --set featureGates.zstdResourceTracker=true .PHONY: e2e-setup-core e2e-setup-core: e2e-setup-core-pre-hook e2e-setup-core-wo-auth e2e-setup-core-post-hook @@ -27,14 +53,35 @@ e2e-setup-core-auth: e2e-setup-core-pre-hook e2e-setup-core-w-auth e2e-setup-cor .PHONY: setup-runtime-e2e-cluster setup-runtime-e2e-cluster: - helm upgrade --install --create-namespace --namespace vela-system --kubeconfig=$(RUNTIME_CLUSTER_CONFIG) --set image.pullPolicy=IfNotPresent --set image.repository=vela-runtime-rollout-test --set image.tag=$(GIT_COMMIT) --wait vela-rollout ./runtime/rollout/charts + helm upgrade --install \ + --create-namespace \ + --namespace vela-system \ + --kubeconfig=$(RUNTIME_CLUSTER_CONFIG) \ + --set image.pullPolicy=IfNotPresent \ + --set image.repository=vela-runtime-rollout-test \ + --set image.tag=$(GIT_COMMIT) \ + --wait vela-rollout \ + ./runtime/rollout/charts .PHONY: e2e-setup e2e-setup: helm install kruise https://github.com/openkruise/charts/releases/download/kruise-1.1.0/kruise-1.1.0.tgz --set featureGates="PreDownloadImageForInPlaceUpdate=true" sh ./hack/e2e/modify_charts.sh - helm upgrade --install --create-namespace --namespace vela-system --set image.pullPolicy=IfNotPresent --set image.repository=vela-core-test --set applicationRevisionLimit=5 --set dependCheckWait=10s --set image.tag=$(GIT_COMMIT) --wait kubevela ./charts/vela-core - helm upgrade --install --namespace vela-system --wait oam-rollout --set image.repository=vela-runtime-rollout-test --set image.tag=$(GIT_COMMIT) ./runtime/rollout/charts + helm upgrade --install \ + --create-namespace \ + --namespace vela-system \ + --set image.pullPolicy=IfNotPresent \ + --set image.repository=vela-core-test \ + --set applicationRevisionLimit=5 \ + --set dependCheckWait=10s \ + --set image.tag=$(GIT_COMMIT) \ + --wait kubevela ./charts/vela-core + helm upgrade --install \ + --namespace vela-system \ + --wait oam-rollout \ + --set image.repository=vela-runtime-rollout-test \ + --set image.tag=$(GIT_COMMIT) \ + ./runtime/rollout/charts go run ./e2e/addon/mock & sleep 15 diff --git a/pkg/features/controller_features.go b/pkg/features/controller_features.go index e86882ab2..cd8af2b07 100644 --- a/pkg/features/controller_features.go +++ b/pkg/features/controller_features.go @@ -58,10 +58,18 @@ const ( // AuthenticateApplication enable the authentication for application AuthenticateApplication featuregate.Feature = "AuthenticateApplication" - // GzipResourceTracker enable the gzip compression for ResourceTracker. It can be useful if you have large + // GzipResourceTracker enables the gzip compression for ResourceTracker. It can be useful if you have large // application that needs to dispatch lots of resources or large resources (like CRD or huge ConfigMap), // which at the cost of slower processing speed due to the extra overhead for compression and decompression. GzipResourceTracker featuregate.Feature = "GzipResourceTracker" + // ZstdResourceTracker enables the zstd compression for ResourceTracker. + // Refer to GzipResourceTracker for its use-cases. It is much faster and more + // efficient than gzip, about 2x faster and compresses to smaller size. + // If you are dealing with very large ResourceTrackers (1MB or so), it should + // have almost NO performance penalties compared to no compression at all. + // If dealing with smaller ResourceTrackers (10KB - 1MB), the performance + // penalties are minimal. + ZstdResourceTracker featuregate.Feature = "ZstdResourceTracker" ) var defaultFeatureGates = map[featuregate.Feature]featuregate.FeatureSpec{ @@ -76,6 +84,7 @@ var defaultFeatureGates = map[featuregate.Feature]featuregate.FeatureSpec{ ApplyResourceByUpdate: {Default: false, PreRelease: featuregate.Alpha}, AuthenticateApplication: {Default: false, PreRelease: featuregate.Alpha}, GzipResourceTracker: {Default: false, PreRelease: featuregate.Alpha}, + ZstdResourceTracker: {Default: false, PreRelease: featuregate.Alpha}, } func init() { diff --git a/pkg/resourcetracker/app.go b/pkg/resourcetracker/app.go index da8ee298d..561844a24 100644 --- a/pkg/resourcetracker/app.go +++ b/pkg/resourcetracker/app.go @@ -95,6 +95,10 @@ func createResourceTracker(ctx context.Context, cli client.Client, app *v1beta1. if utilfeature.DefaultMutableFeatureGate.Enabled(features.GzipResourceTracker) { rt.Spec.Compression.Type = compression.Gzip } + // zstd compressor will have higher priority when both gzip and zstd are enabled. + if utilfeature.DefaultMutableFeatureGate.Enabled(features.ZstdResourceTracker) { + rt.Spec.Compression.Type = compression.Zstd + } if err := cli.Create(ctx, rt); err != nil { return nil, err } diff --git a/pkg/utils/compression/types.go b/pkg/utils/compression/types.go index f49640f77..5ad2ee3b0 100644 --- a/pkg/utils/compression/types.go +++ b/pkg/utils/compression/types.go @@ -20,8 +20,10 @@ package compression type Type string const ( - // Uncompressed . + // Uncompressed does not compress or encode data Uncompressed Type = "" - // Gzip . + // Gzip compresses data using gzip and encodes it using base64 Gzip Type = "gzip" + // Zstd compresses data using zstd and encodes it using base64 + Zstd Type = "zstd" ) diff --git a/pkg/utils/compression/zstd.go b/pkg/utils/compression/zstd.go new file mode 100644 index 000000000..7e12de86a --- /dev/null +++ b/pkg/utils/compression/zstd.go @@ -0,0 +1,110 @@ +/* +Copyright 2022 The KubeVela Authors. + +Licensed 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 compression + +import ( + "encoding/base64" + "encoding/json" + + "github.com/klauspost/compress/zstd" +) + +// Create a writer that caches compressors. For this operation type we supply a nil Reader. +var encoder, _ = zstd.NewWriter(nil, + // We use the default levels here because we got pretty good results. + // It is almost as fast as no compression at all when the object is large enough. + // Even with small objects, it is still very fast and efficient. + // + // Tests are here: /apis/core.oam.dev/v1beta1/resourcetracker_types_test.go + // + // Here are results: + // zstd.SpeedFastest: + // Compressed Size: + // uncompressed: 2131455 bytes 100.00% + // gzip: 273057 bytes 12.81% + // zstd: 191737 bytes 9.00% + // Marshal Time: + // no compression: 37740514 ns 1.00x + // gzip: 97389702 ns 2.58x + // zstd: 39866808 ns 1.06x + // zstd.SpeedDefault: + // Compressed Size: + // uncompressed: 2131455 bytes 100.00% + // gzip: 273057 bytes 12.81% + // zstd: 171577 bytes 8.05% + // Marshal Time: + // no compression: 42272142 ns 1.00x + // gzip: 90474722 ns 2.14x + // zstd: 39070416 ns 0.92x + // zstd.SpeedBetterCompression: + // Compressed Size: + // uncompressed: 2131455 bytes 100.00% + // gzip: 273057 bytes 12.81% + // zstd: 149061 bytes 6.99% + // Marshal Time: + // no compression: 38826717 ns 1.00x + // gzip: 94855264 ns 2.44x + // zstd: 48524197 ns 1.25x + zstd.WithEncoderLevel(zstd.SpeedDefault), + // TODO(charlie0129): give a dictionary to compressor to get even more improvements. + // + // Since we are dealing with highly-specialized small JSON data, a dictionary will + // give massive improvements, around 3x both (de)compression speed and size reduction, + // according to Facebook https://github.com/facebook/zstd#the-case-for-small-data-compression. + // zstd.WithEncoderDict(), +) + +// Create a reader that caches decompressors. +var decoder, _ = zstd.NewReader(nil) + +func compress(src []byte) []byte { + return encoder.EncodeAll(src, make([]byte, 0, len(src))) +} + +func decompress(src []byte) ([]byte, error) { + return decoder.DecodeAll(src, nil) +} + +// ZstdObjectToString marshals the object into json, compress it with zstd, +// encode the result with base64. +func ZstdObjectToString(obj interface{}) (string, error) { + bs, err := json.Marshal(obj) + if err != nil { + return "", err + } + + compressedBytes := compress(bs) + + return base64.StdEncoding.EncodeToString(compressedBytes), nil +} + +// UnZstdStringToObject decodes the compressed string with base64, +// decompresses it with zstd, and unmarshals it. obj must be a pointer so that +// it can be updated. +func UnZstdStringToObject(encoded string, obj interface{}) error { + decoded, err := base64.StdEncoding.DecodeString(encoded) + if err != nil { + return err + } + + decompressed, err := decompress(decoded) + if err != nil { + return err + } + + return json.Unmarshal(decompressed, obj) +} diff --git a/pkg/utils/compression/zstd_test.go b/pkg/utils/compression/zstd_test.go new file mode 100644 index 000000000..c0b38b0ad --- /dev/null +++ b/pkg/utils/compression/zstd_test.go @@ -0,0 +1,50 @@ +/* +Copyright 2022 The KubeVela Authors. + +Licensed 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 compression + +import ( + "math" + "testing" + + "github.com/stretchr/testify/assert" + v1 "k8s.io/api/core/v1" +) + +func TestZstdCompression(t *testing.T) { + obj := v1.ConfigMap{ + Data: map[string]string{"1234": "5678"}, + } + + str, err := ZstdObjectToString(obj) + assert.NoError(t, err) + objOut := v1.ConfigMap{} + err = UnZstdStringToObject(str, &objOut) + assert.NoError(t, err) + assert.Equal(t, obj, objOut) + + // Invalid obj + _, err = ZstdObjectToString(math.Inf(1)) + assert.Error(t, err) + + // Invalid base64 string + err = UnZstdStringToObject(".dew;.3234", &objOut) + assert.Error(t, err) + + // Invalid zstd binary data + err = UnZstdStringToObject("MTIzNDUK", &objOut) + assert.Error(t, err) +}