Client DataStore data compression

pkg/client/client.go

@@ -2,7 +2,9 @@ package client
 
 import (
 	"bytes"
+	"crypto/sha256"
 	"encoding/base64"
+	"encoding/hex"
 	"encoding/json"
 	"fmt"
 	"io"
@@ -30,6 +32,13 @@ type TelemetryAuth struct {
 	IssueDate types.TelemetryTimeStamp `json:"issueDate"`
 }
 
+func (t *TelemetryAuth) HashToken(token types.TelemetryAuthToken) {
+	hash := sha256.New()
+	hash.Write([]byte(token))
+	hashedToken := hex.EncodeToString(hash.Sum(nil))
+	t.Token = types.TelemetryAuthToken(hashedToken)
+}
+
 type TelemetryClient struct {
 	cfg       *config.Config
 	auth      TelemetryAuth
@@ -38,6 +47,7 @@ type TelemetryClient struct {
 
 func NewTelemetryClient(cfg *config.Config) (tc *TelemetryClient, err error) {
 	tc = &TelemetryClient{cfg: cfg}
+	tc.auth.HashToken(tc.auth.Token)
 	tc.processor, err = telemetrylib.NewTelemetryProcessor(&cfg.DataStores)
 	return
 }

pkg/lib/processor.go

@@ -1,6 +1,8 @@
 package telemetrylib
 
 import (
+	"bytes"
+	"compress/gzip"
 	"fmt"
 	"log"
 	"strings"
@@ -20,6 +22,12 @@ type TelemetryProcessor interface {
 		tags types.Tags,
 	) (err error)
 
+	// Data compression method
+	CompressData(
+		data []byte,
+		format string,
+	) (compressedData []byte, err error)
+
 	// Generate telemetry bundle
 	GenerateBundle(
 		clientId int64,
@@ -107,7 +115,11 @@ func NewTelemetryProcessor(cfg *config.DBConfig) (TelemetryProcessor, error) {
 }
 
 func (p *TelemetryProcessorImpl) AddData(telemetry types.TelemetryType, marshaledData []byte, tags types.Tags) (err error) {
-	dataItemRow, err := NewTelemetryDataItemRow(telemetry, tags, marshaledData)
+	compressedData, err := p.CompressData(marshaledData, "gzip")
+	if err != nil {
+		return fmt.Errorf("unable to compress telemetry data: %s", err.Error())
+	}
+	dataItemRow, err := NewTelemetryDataItemRow(telemetry, tags, compressedData)
 	if err != nil {
 		return fmt.Errorf("unable to create telemetry data: %s", err.Error())
 	}
@@ -116,6 +128,37 @@ func (p *TelemetryProcessorImpl) AddData(telemetry types.TelemetryType, marshale
 	return
 }
 
+func (p *TelemetryProcessorImpl) CompressData(marshaledData []byte, format string) (compressedData []byte, err error) {
+	switch format {
+	case "gzip":
+		zd, err := p.compressGZIP(marshaledData)
+		return zd.Bytes(), err
+	}
+	return
+}
+
+func (p *TelemetryProcessorImpl) compressGZIP(marshaledData []byte) (compressedData bytes.Buffer, err error) {
+	var encoder *gzip.Writer
+	var tmpBuffer bytes.Buffer
+
+	encoder, err = gzip.NewWriterLevel(&tmpBuffer, gzip.BestCompression)
+	if err != nil {
+		return tmpBuffer, err
+	}
+
+	_, err = encoder.Write(marshaledData)
+	if err != nil {
+		return tmpBuffer, err
+	}
+	defer encoder.Close()
+
+	if err := encoder.Close(); err != nil {
+		return tmpBuffer, err
+	}
+
+	return tmpBuffer, nil
+}
+
 func (p *TelemetryProcessorImpl) GenerateBundle(clientId int64, customerId string, tags types.Tags) (bundleRow *TelemetryBundleRow, err error) {
 
 	bundleRow, err = NewTelemetryBundleRow(clientId, customerId, tags)