feat(strm-1882): add micro-aggregation batch job & FieldMetaData support

go.mod

@@ -14,6 +14,7 @@ require (
 	github.com/int128/oauth2cli v1.14.0
 	github.com/jedib0t/go-pretty/v6 v6.4.0
 	github.com/joho/godotenv v1.4.0
+	github.com/lithammer/dedent v1.1.0
 	github.com/magiconair/properties v1.8.6
 	github.com/manifoldco/promptui v0.9.0
 	github.com/pkg/browser v0.0.0-20210911075715-681adbf594b8
@@ -23,7 +24,7 @@ require (
 	github.com/spf13/cobra v1.6.1
 	github.com/spf13/pflag v1.0.5
 	github.com/spf13/viper v1.13.0
-	github.com/strmprivacy/api-definitions-go/v2 v2.59.0
+	github.com/strmprivacy/api-definitions-go/v2 v2.65.1
 	golang.org/x/exp v0.0.0-20221026153819-32f3d567a233 // indirect
 	golang.org/x/oauth2 v0.1.0
 	golang.org/x/sync v0.1.0
@@ -33,8 +34,6 @@ require (
 	sigs.k8s.io/yaml v1.3.0
 )
 
-require github.com/lithammer/dedent v1.1.0
-
 require (
 	github.com/chzyer/readline v0.0.0-20180603132655-2972be24d48e // indirect
 	github.com/cpuguy83/go-md2man/v2 v2.0.2 // indirect

go.sum

@@ -271,8 +271,8 @@ github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/
 github.com/stretchr/testify v1.7.4/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
 github.com/stretchr/testify v1.8.0 h1:pSgiaMZlXftHpm5L7V1+rVB+AZJydKsMxsQBIJw4PKk=
 github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
-github.com/strmprivacy/api-definitions-go/v2 v2.59.0 h1:Q/GO85xbGEYZio8dQYneghtR8jF8zMo92PrTJVx+MG4=
-github.com/strmprivacy/api-definitions-go/v2 v2.59.0/go.mod h1:3uFjMuBEQSzrRQzaKEgIrLbBWRdya9DTYCQZqyS7nEw=
+github.com/strmprivacy/api-definitions-go/v2 v2.65.1 h1:oA6/BznvOV6MDoXjiwgZuN0EpMvNabrCzJhjQNRf9ak=
+github.com/strmprivacy/api-definitions-go/v2 v2.65.1/go.mod h1:PdBDOOXTlTzATGv1M3Hf0cNnXmq4zdtg5eNcwYHvOkM=
 github.com/subosito/gotenv v1.4.1 h1:jyEFiXpy21Wm81FBN71l9VoMMV8H8jG+qIK3GCpY6Qs=
 github.com/subosito/gotenv v1.4.1/go.mod h1:ayKnFf/c6rvx/2iiLrJUk1e6plDbT3edrFNGqEflhK0=
 github.com/thoas/go-funk v0.9.1 h1:O549iLZqPpTUQ10ykd26sZhzD+rmR5pWhuElrhbC20M=
@@ -598,6 +598,7 @@ google.golang.org/grpc v1.34.0/go.mod h1:WotjhfgOW/POjDeRt8vscBtXq+2VjORFy659qA5
 google.golang.org/grpc v1.35.0/go.mod h1:qjiiYl8FncCW8feJPdyg3v6XW24KsRHe+dy9BAGRRjU=
 google.golang.org/grpc v1.36.0/go.mod h1:qjiiYl8FncCW8feJPdyg3v6XW24KsRHe+dy9BAGRRjU=
 google.golang.org/grpc v1.46.0/go.mod h1:vN9eftEi1UMyUsIF80+uQXhHjbXYbm0uXoFCACuMGWk=
+google.golang.org/grpc v1.50.0/go.mod h1:ZgQEeidpAuNRZ8iRrlBKXZQP1ghovWIVhdJRyCDK+GI=
 google.golang.org/grpc v1.50.1 h1:DS/BukOZWp8s6p4Dt/tOaJaTQyPyOoCcrjroHuCeLzY=
 google.golang.org/grpc v1.50.1/go.mod h1:ZgQEeidpAuNRZ8iRrlBKXZQP1ghovWIVhdJRyCDK+GI=
 google.golang.org/protobuf v0.0.0-20200109180630-ec00e32a8dfd/go.mod h1:DFci5gLYBciE7Vtevhsrf46CRTquxDuWsQurQQe4oz8=

pkg/auth/stored_token.go

@@ -7,7 +7,6 @@ import (
 	"fmt"
 	log "github.com/sirupsen/logrus"
 	"golang.org/x/oauth2"
-	"io/ioutil"
 	"net/url"
 	"os"
 	"path"
@@ -36,7 +35,7 @@ func (e *EmptyTokenError) Error() string {
 
 func (authenticator *Authenticator) LoadLogin() error {
 	filename := authenticator.getSaveFilename()
-	b, err := ioutil.ReadFile(filename)
+	b, err := os.ReadFile(filename)
 
 	if errors.Is(err, os.ErrNotExist) {
 		return common.UnauthenticatedError()
@@ -65,7 +64,7 @@ func (authenticator *Authenticator) storeLogin() string {
 	common.CliExit(err)
 	b, err := json.Marshal(authenticator.storedToken)
 	common.CliExit(err)
-	err = ioutil.WriteFile(filename, b, 0644)
+	err = os.WriteFile(filename, b, 0644)
 	common.CliExit(err)
 	return filename
 }

pkg/common/common.go

@@ -12,7 +12,6 @@ import (
 	"google.golang.org/grpc/metadata"
 	"google.golang.org/grpc/status"
 	"gopkg.in/natefinch/lumberjack.v2"
-	"io/ioutil"
 	"os"
 	"path"
 	"runtime"
@@ -123,7 +122,7 @@ func MarkRequiredFlags(cmd *cobra.Command, flagNames ...string) {
 func GetActiveProject() string {
 	activeProjectFilePath := path.Join(ConfigPath, activeProjectFilename)
 
-	bytes, err := ioutil.ReadFile(activeProjectFilePath)
+	bytes, err := os.ReadFile(activeProjectFilePath)
 	CliExit(err)
 	activeProject := string(bytes)
 	log.Infoln("Current active project is: " + activeProject)

pkg/context/context.go

@@ -2,7 +2,7 @@ package context
 
 import (
 	"fmt"
-	"io/ioutil"
+	"os"
 	"path"
 	"strings"
 	"strmprivacy/strm/pkg/auth"
@@ -25,7 +25,7 @@ type apiUrls struct {
 
 func showConfiguration() {
 	configFilepath := findConfigFile()
-	contents, err := ioutil.ReadFile(configFilepath)
+	contents, err := os.ReadFile(configFilepath)
 	common.CliExit(err)
 
 	configuration := configuration{
@@ -48,7 +48,7 @@ type savedEntity struct {
 
 func entityInfo(args []string) {
 	filepath := path.Join(common.ConfigPath, common.SavedEntitiesDirectory, args[0]+".json")
-	contents, err := ioutil.ReadFile(filepath)
+	contents, err := os.ReadFile(filepath)
 	common.CliExit(err)
 
 	entity := savedEntity{Path: filepath, Contents: contents}
@@ -61,7 +61,7 @@ func showAccountDetails() {
 }
 
 func listSavedEntities(p string) []string {
-	files, err := ioutil.ReadDir(p)
+	dirEntries, err := os.ReadDir(p)
 
 	if err != nil {
 		return []string{}
@@ -70,26 +70,26 @@ func listSavedEntities(p string) []string {
 	entityTypeTemplate := path.Base(p) + "/%v"
 	var entities = make([]string, 0)
 
-	for _, f := range files {
-		if f.IsDir() {
-			entities = append(entities, listSavedEntities(path.Join(p, f.Name()))...)
-		} else if strings.HasSuffix(f.Name(), "json") && !strings.HasPrefix(f.Name(), auth.StrmCredsFilePrefix) {
-			entities = append(entities, fmt.Sprintf(entityTypeTemplate, strings.Replace(f.Name(), ".json", "", -1)))
+	for _, dirEntry := range dirEntries {
+		if dirEntry.IsDir() {
+			entities = append(entities, listSavedEntities(path.Join(p, dirEntry.Name()))...)
+		} else if strings.HasSuffix(dirEntry.Name(), "json") && !strings.HasPrefix(dirEntry.Name(), auth.StrmCredsFilePrefix) {
+			entities = append(entities, fmt.Sprintf(entityTypeTemplate, strings.Replace(dirEntry.Name(), ".json", "", -1)))
 		}
 	}
 
 	return entities
 }
 
 func findConfigFile() string {
-	files, err := ioutil.ReadDir(common.ConfigPath)
+	dirEntries, err := os.ReadDir(common.ConfigPath)
 	common.CliExit(err)
 
 	var configFilepath string
 
-	for _, f := range files {
-		if f.Name() == common.DefaultConfigFilename+common.DefaultConfigFileSuffix || f.Name() == common.DefaultConfigFilename+common.DefaultConfigFileSuffix {
-			configFilepath = path.Join(common.ConfigPath, f.Name())
+	for _, dirEntry := range dirEntries {
+		if dirEntry.Name() == common.DefaultConfigFilename+common.DefaultConfigFileSuffix || dirEntry.Name() == common.DefaultConfigFilename+common.DefaultConfigFileSuffix {
+			configFilepath = path.Join(common.ConfigPath, dirEntry.Name())
 		}
 	}
 

pkg/context/project.go

@@ -5,7 +5,6 @@ import (
 	"fmt"
 	log "github.com/sirupsen/logrus"
 	"github.com/spf13/pflag"
-	"io/ioutil"
 	"os"
 	"path"
 	"strmprivacy/strm/pkg/common"
@@ -72,7 +71,7 @@ func initActiveProject() {
 func saveActiveProject(projectName string) {
 	activeProjectFilepath := path.Join(common.ConfigPath, activeProjectFilename)
 
-	err := ioutil.WriteFile(
+	err := os.WriteFile(
 		activeProjectFilepath,
 		[]byte(projectName),
 		0644,

pkg/entity/batch_job/batch_job.go

@@ -3,12 +3,15 @@ package batch_job
 import (
 	"bytes"
 	"context"
+	"errors"
+	"fmt"
 	"github.com/golang/protobuf/jsonpb"
 	"github.com/spf13/cobra"
+	"github.com/spf13/pflag"
 	"github.com/strmprivacy/api-definitions-go/v2/api/batch_jobs/v1"
 	"github.com/strmprivacy/api-definitions-go/v2/api/entities/v1"
 	"google.golang.org/grpc"
-	"io/ioutil"
+	"os"
 	"strings"
 	"strmprivacy/strm/pkg/common"
 	"strmprivacy/strm/pkg/entity/policy"
@@ -19,6 +22,11 @@ import (
 var client batch_jobs.BatchJobsServiceClient
 var apiContext context.Context
 
+type refWithStates struct {
+	ref    *entities.BatchJobRef
+	states []*entities.BatchJobState
+}
+
 func SetupClient(clientConnection *grpc.ClientConn, ctx context.Context) {
 	apiContext = ctx
 	client = batch_jobs.NewBatchJobsServiceClient(clientConnection)
@@ -59,15 +67,39 @@ func del(id *string, cmd *cobra.Command) {
 
 func create(cmd *cobra.Command) {
 	flags := cmd.Flags()
-	batchJobFile := util.GetStringAndErr(flags, batchJobsFileFlagName)
+	batchJobFile := util.GetStringAndErr(flags, batchJobFileFlagName)
+	batchJobType := getBatchJobType(flags)
 
-	batchJobData, err := ioutil.ReadFile(batchJobFile)
+	batchJobData, err := os.ReadFile(batchJobFile)
 	if err != nil {
 		common.CliExit(err)
 	}
 
+	var batchJobWrapper *entities.BatchJobWrapper
+	if batchJobType == encryptionType {
+		batchJobWrapper = createEncryptionBatchJob(cmd, batchJobData, flags)
+	} else if batchJobType == microAggregationType {
+		batchJobWrapper = createMicroAggregationBatchJob(cmd, batchJobData)
+	}
+	createBatchJobRequest := &batch_jobs.CreateBatchJobRequest{Job: batchJobWrapper}
+	response, err := client.CreateBatchJob(apiContext, createBatchJobRequest)
+	common.CliExit(err)
+
+	printer.Print(response)
+}
+
+func getBatchJobType(flags *pflag.FlagSet) string {
+	batchJobType := util.GetStringAndErr(flags, batchJobTypeFlagName)
+	if !(batchJobType == encryptionType || batchJobType == microAggregationType) {
+		common.CliExit(errors.New(fmt.Sprintf("Batch job type should be one of: %s, %s",
+			encryptionType, microAggregationType)))
+	}
+	return batchJobType
+}
+
+func createEncryptionBatchJob(cmd *cobra.Command,batchJobData []byte, flags *pflag.FlagSet) *entities.BatchJobWrapper {
 	batchJob := &entities.BatchJob{}
-	err = jsonpb.Unmarshal(bytes.NewReader(batchJobData), batchJob)
+	err := jsonpb.Unmarshal(bytes.NewReader(batchJobData), batchJob)
 	if err != nil {
 		common.CliExit(err)
 	}
@@ -76,15 +108,15 @@ func create(cmd *cobra.Command) {
 	if policyId != "" {
 		batchJob.PolicyId = policyId
 	}
-	setCommonProjectIds(batchJob, projectId)
-	createBatchJobRequest := &batch_jobs.CreateBatchJobRequest{BatchJob: batchJob}
-	response, err := client.CreateBatchJob(apiContext, createBatchJobRequest)
-	common.CliExit(err)
-
-	printer.Print(response)
+	setEncryptionBatchJobProjectIds(batchJob, projectId)
+	return &entities.BatchJobWrapper{
+		Job: &entities.BatchJobWrapper_EncryptionBatchJob{
+			EncryptionBatchJob: batchJob,
+		},
+	}
 }
 
-func setCommonProjectIds(batchJob *entities.BatchJob, projectId string) {
+func setEncryptionBatchJobProjectIds(batchJob *entities.BatchJob, projectId string) {
 	if batchJob.Ref == nil {
 		// normal situation where the whole ref attribute in the json is absent.
 		batchJob.Ref = &entities.BatchJobRef{}
@@ -96,7 +128,31 @@ func setCommonProjectIds(batchJob *entities.BatchJob, projectId string) {
 	for _, d := range batchJob.DerivedData {
 		d.Target.DataConnectorRef.ProjectId = projectId
 	}
+}
 
+func createMicroAggregationBatchJob(cmd *cobra.Command, data []byte) *entities.BatchJobWrapper {
+	batchJob := &entities.MicroAggregationBatchJob{}
+	err := jsonpb.Unmarshal(bytes.NewReader(data), batchJob)
+	if err != nil {
+		common.CliExit(err)
+	}
+	projectId := project.GetProjectId(cmd)
+	setMicroAggregationBatchJobProjectIds(batchJob, projectId)
+	return &entities.BatchJobWrapper{
+		Job: &entities.BatchJobWrapper_MicroAggregationBatchJob{
+			MicroAggregationBatchJob: batchJob,
+		},
+	}
+}
+
+func setMicroAggregationBatchJobProjectIds(batchJob *entities.MicroAggregationBatchJob, projectId string) {
+	if batchJob.Ref == nil {
+		// normal situation where the whole ref attribute in the json is absent.
+		batchJob.Ref = &entities.BatchJobRef{}
+	}
+	batchJob.Ref.ProjectId = projectId
+	batchJob.SourceData.DataConnectorRef.ProjectId = projectId
+	batchJob.TargetData.DataConnectorRef.ProjectId = projectId
 }
 
 func namesCompletion(cmd *cobra.Command, args []string, complete string) ([]string, cobra.ShellCompDirective) {
@@ -119,3 +175,19 @@ func namesCompletion(cmd *cobra.Command, args []string, complete string) ([]stri
 	}
 	return batchJobIds, cobra.ShellCompDirectiveNoFileComp
 }
+
+func toRefWithStates(batchJob *entities.BatchJobWrapper) refWithStates {
+	var ref *entities.BatchJobRef
+	var states []*entities.BatchJobState
+	if encryptionBatchJob := batchJob.GetEncryptionBatchJob(); encryptionBatchJob != nil {
+		ref = encryptionBatchJob.Ref
+		states = encryptionBatchJob.States
+	} else if microAggregationBatchJob := batchJob.GetMicroAggregationBatchJob(); microAggregationBatchJob != nil {
+		ref = microAggregationBatchJob.Ref
+		states = microAggregationBatchJob.States
+	}
+	return refWithStates{
+		states: states,
+		ref:    ref,
+	}
+}

pkg/entity/batch_job/cmd.go

@@ -2,18 +2,20 @@ package batch_job
 
 import (
 	"github.com/spf13/cobra"
-	"strmprivacy/strm/pkg/common"
 	"strmprivacy/strm/pkg/entity/policy"
 )
 
 const (
-	batchJobsFileFlagName = "file"
+	batchJobFileFlagName = "file"
+	batchJobTypeFlagName = "type"
+	encryptionType = "encryption"
+	microAggregationType = "micro-aggregation"
 )
 
 func DeleteCmd() *cobra.Command {
 	batchJob := &cobra.Command{
 		Use:               "batch-job (id ...)",
-		Short:             "Delete on or more Batch Jobs by id",
+		Short:             "Delete one or more Batch Jobs by id",
 		DisableAutoGenTag: true,
 		PreRun: func(cmd *cobra.Command, args []string) {
 			printer = configurePrinter(cmd)
@@ -78,11 +80,12 @@ func CreateCmd() *cobra.Command {
 
 	flags := batchJob.Flags()
 
-	flags.StringP(batchJobsFileFlagName, "F", "",
+	flags.StringP(batchJobFileFlagName, "F", "",
 		`the path to the JSON file containing the batch job configuration`)
+	flags.StringP(batchJobTypeFlagName, "T", "encryption",
+		`the type of batch job (encryption, micro-aggregation), defaults to encryption`)
 	policy.SetupFlags(batchJob, flags)
-	err := batchJob.MarkFlagRequired(batchJobsFileFlagName)
-	common.CliExit(err)
+	_ = batchJob.MarkFlagRequired(batchJobFileFlagName)
 
 	return batchJob
 }