Skip to content

Commit e0f9a44

Browse files
ricardopiedadericardopiedade
authored
Merge pull request #29 from miniclip/feature/transaction_batch_op_indexes
Feature/transaction batch op indexes
2 parents bd1d453 + bb68bef commit e0f9a44

File tree

5 files changed

+88
-59
lines changed

5 files changed

+88
-59
lines changed

‎go.mod‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,6 @@ require (
1919
github.com/onsi/gomega v1.10.1
2020
github.com/pelletier/go-buffruneio v0.2.0 // indirect
2121
github.com/pkg/errors v0.9.1 // indirect
22-
github.com/pmezard/go-difflib v1.0.0 // indirect
2322
github.com/sergi/go-diff v1.0.0 // indirect
2423
github.com/src-d/gcfg v1.3.0 // indirect
2524
github.com/stretchr/objx v0.1.1 // indirect
@@ -29,5 +28,5 @@ require (
2928
gopkg.in/src-d/go-git-fixtures.v3 v3.5.0 // indirect
3029
gopkg.in/src-d/go-git.v4 v4.4.1
3130
gopkg.in/warnings.v0 v0.1.2 // indirect
32-
gopkg.in/yaml.v2 v2.3.0
31+
gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b
3332
)

‎go.sum‎

Lines changed: 2 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,6 @@ github.com/anmitsu/go-shlex v0.0.0-20200514113438-38f4b401e2be h1:9AeTilPcZAjCFI
44
github.com/anmitsu/go-shlex v0.0.0-20200514113438-38f4b401e2be/go.mod h1:ySMOLuWl6zY27l47sB3qLNK6tF2fkHG55UZxx8oIVo4=
55
github.com/cbroglie/mustache v1.0.1 h1:ivMg8MguXq/rrz2eu3tw6g3b16+PQhoTn6EZAhst2mw=
66
github.com/cbroglie/mustache v1.0.1/go.mod h1:R/RUa+SobQ14qkP4jtx5Vke5sDytONDQXNLPY/PO69g=
7-
github.com/davecgh/go-spew v1.1.0 h1:ZDRjVQ15GmhC3fiQ8ni8+OwkZQO4DARzQgrnXU1Liz8=
87
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
98
github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
109
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
@@ -22,7 +21,6 @@ github.com/golang/protobuf v1.4.0-rc.1.0.20200221234624-67d41d38c208/go.mod h1:x
2221
github.com/golang/protobuf v1.4.0-rc.2/go.mod h1:LlEzMj4AhA7rCAGe4KMBDvJI+AwstrUpVNzEA03Pprs=
2322
github.com/golang/protobuf v1.4.0-rc.4.0.20200313231945-b860323f09d0/go.mod h1:WU3c8KckQ9AFe+yFwt9sWVRKCVIyN9cPHBJSNnbL67w=
2423
github.com/golang/protobuf v1.4.0/go.mod h1:jodUvKwWbYaEsadDk5Fwe5c77LiNKVO9IDvqG2KuDX0=
25-
github.com/golang/protobuf v1.4.2 h1:+Z5KGCizgyZCbGh1KZqA0fcLLkwbsjIzS4aV2v7wJX0=
2624
github.com/golang/protobuf v1.4.2/go.mod h1:oDoupMAO8OvCJWAcko0GGGIgR6R6ocIYbsSw735rRwI=
2725
github.com/google/go-cmp v0.3.0/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=
2826
github.com/google/go-cmp v0.3.1/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=
@@ -49,8 +47,6 @@ github.com/onsi/ginkgo v1.6.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+W
4947
github.com/onsi/ginkgo v1.12.1/go.mod h1:zj2OWP4+oCPe1qIXoGWkgMRwljMUYCdkwsT2108oapk=
5048
github.com/onsi/ginkgo v1.16.1 h1:foqVmeWDD6yYpK+Yz3fHyNIxFYNxswxqNFjSKe+vI54=
5149
github.com/onsi/ginkgo v1.16.1/go.mod h1:CObGmKUOKaSC0RjmoAK7tKyn4Azo5P2IWuoMnvwxz1E=
52-
github.com/onsi/gomega v1.4.0 h1:p/ZBjQI9G/VwoPrslo/sqS6R5vHU9Od60+axIiP6WuQ=
53-
github.com/onsi/gomega v1.4.0/go.mod h1:C1qb7wdrVGGVU+Z6iS04AVkA3Q65CEZX59MT0QO5uiA=
5450
github.com/onsi/gomega v1.7.1/go.mod h1:XdKZgCCFLUoM/7CFJVPcG8C1xQ1AJ0vpAezJrB7JYyY=
5551
github.com/onsi/gomega v1.10.1 h1:o0+MgICZLuZ7xjH7Vx6zS/zcu93/BEp1VwkIW1mEXCE=
5652
github.com/onsi/gomega v1.10.1/go.mod h1:iN09h71vgCQne3DLsj+A5owkum+a2tYe+TOCB1ybHNo=
@@ -67,22 +63,16 @@ github.com/src-d/gcfg v1.3.0/go.mod h1:p/UMsR43ujA89BJY9duynAwIpvqEujIH/jFlfL7jW
6763
github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
6864
github.com/stretchr/objx v0.1.1 h1:2vfRuCMp5sSVIDSqO8oNnWJq7mPa6KVP3iPIwFBuy8A=
6965
github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
70-
github.com/stretchr/testify v1.2.2 h1:bSDNvY7ZPG5RlJ8otE/7V6gMiyenm9RtJ7IUVIAoJ1w=
71-
github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
7266
github.com/stretchr/testify v1.5.1 h1:nOGnQDM7FYENwehXlg/kFVnos3rEvtKTjRvOWSzb6H4=
7367
github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA=
7468
github.com/xanzy/ssh-agent v0.1.0 h1:lOhdXLxtmYjaHc76ZtNmJWPg948y/RnT+3N3cvKWFzY=
7569
github.com/xanzy/ssh-agent v0.1.0/go.mod h1:0NyE30eGUDliuLEHJgYte/zncp2zdTStcOnWhgSqHD8=
7670
github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
77-
golang.org/x/crypto v0.0.0-20180621125126-a49355c7e3f8 h1:h7zdf0RiEvWbYBKIx4b+q41xoUVnMmvsGZnIVE5syG8=
78-
golang.org/x/crypto v0.0.0-20180621125126-a49355c7e3f8/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
7971
golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
8072
golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
8173
golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9 h1:psW17arqaxU48Z5kZ0CQnkZWQJsqcURM6tKiBApRjXI=
8274
golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
8375
golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
84-
golang.org/x/net v0.0.0-20180629035331-4cb1c02c05b0 h1:eOjEPieBzQ+rKOvQTqwbkm/0BdWz2JQwUzaa97tcZ8k=
85-
golang.org/x/net v0.0.0-20180629035331-4cb1c02c05b0/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
8676
golang.org/x/net v0.0.0-20180906233101-161cd47e91fd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
8777
golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
8878
golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
@@ -92,8 +82,6 @@ golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwY
9282
golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
9383
golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
9484
golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
95-
golang.org/x/sys v0.0.0-20180627142611-7138fd3d9dc8 h1:RI4LLZfYDSosZMJ7FzhhEQbwo7tA8Bp9Vhml1PukQsg=
96-
golang.org/x/sys v0.0.0-20180627142611-7138fd3d9dc8/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
9785
golang.org/x/sys v0.0.0-20180909124046-d0be0721c37e/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
9886
golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
9987
golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
@@ -104,7 +92,6 @@ golang.org/x/sys v0.0.0-20200323222414-85ca7c5b95cd/go.mod h1:h1NjWce9XRLGQEsW7w
10492
golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
10593
golang.org/x/sys v0.0.0-20210112080510-489259a85091 h1:DMyOG0U+gKfu8JZzg2UQe9MeaC1X+xQWlAKcRnjxjCw=
10694
golang.org/x/sys v0.0.0-20210112080510-489259a85091/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
107-
golang.org/x/text v0.3.0 h1:g61tztE5qeGQ89tm6NTjjM9VPIm088od1l6aSorWRWg=
10895
golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
10996
golang.org/x/text v0.3.3 h1:cokOdA+Jmi5PJGXLlLllQSgYigAEfHXJAERHVMaCc2k=
11097
golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
@@ -121,7 +108,6 @@ google.golang.org/protobuf v0.0.0-20200221191635-4d8936d0db64/go.mod h1:kwYJMbMJ
121108
google.golang.org/protobuf v0.0.0-20200228230310-ab0ca4ff8a60/go.mod h1:cfTl7dwQJ+fmap5saPgwCLgHXTUD7jkjRqWcaiX5VyM=
122109
google.golang.org/protobuf v1.20.1-0.20200309200217-e05f789c0967/go.mod h1:A+miEFZTKqfCUM6K7xSMQL9OKL/b6hQv+e19PK+JZNE=
123110
google.golang.org/protobuf v1.21.0/go.mod h1:47Nbq4nVaFHyn7ilMalzfO3qCViNmqZ2kzikPIcrTAo=
124-
google.golang.org/protobuf v1.23.0 h1:4MY060fB1DLGMB/7MBTLnwQUY6+F09GEiz6SsrNqyzM=
125111
google.golang.org/protobuf v1.23.0/go.mod h1:EGpADcykh3NcUnDUJcl1+ZksZNG86OlYog2l/sGQquU=
126112
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
127113
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
@@ -138,7 +124,7 @@ gopkg.in/warnings.v0 v0.1.2 h1:wFXVbFY8DY5/xOe1ECiWdKCzZlxgshcYVNkBHstARME=
138124
gopkg.in/warnings.v0 v0.1.2/go.mod h1:jksf8JmL6Qr/oQM2OXTHunEvvTAsrWBLb6OOjuVWRNI=
139125
gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
140126
gopkg.in/yaml.v2 v2.2.4/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
141-
gopkg.in/yaml.v2 v2.2.8 h1:obN1ZagJSUGI0Ek/LBmuj4SNLPfIny3KsKFopxRdj10=
142-
gopkg.in/yaml.v2 v2.2.8/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
143127
gopkg.in/yaml.v2 v2.3.0 h1:clyUAQHOM3G0M3f5vQj7LuJrETvjVot3Z5el9nffUtU=
144128
gopkg.in/yaml.v2 v2.3.0/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
129+
gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b h1:h8qDotaEPuJATrMmW04NCwg7v22aHH28wwpauUhK9Oo=
130+
gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

‎internal/importer/consul.go‎

Lines changed: 14 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,8 @@
11
package importer
22

33
import (
4+
"strconv"
5+
46
"github.com/miniclip/gonsul/internal/entities"
57
"github.com/miniclip/gonsul/internal/util"
68

@@ -15,21 +17,23 @@ const consulTxnLimit = 64
1517
const maximumPayloadSize = 500000 // max size is actually 512kb
1618

1719
// processConsulTransaction ...
18-
func (i *importer) processConsulTransaction(transactions []entities.ConsulTxn) {
20+
func (i *importer) processConsulTransaction(transactions []entities.ConsulTxn, batchNumber int) {
21+
batch := strconv.Itoa(batchNumber)
22+
1923
// Encode our transaction into a JSON payload
2024
jsonPayload, err := json.Marshal(transactions)
2125
if err != nil {
22-
util.ExitError(errors.New("Marshal: "+err.Error()), util.ErrorFailedJsonEncode, i.logger)
26+
util.ExitError(errors.New("Marshal: "+err.Error()+" in Batch "+batch), util.ErrorFailedJsonEncode, i.logger)
2327
}
2428

2529
// Create our URL
2630
consulUrl := i.config.GetConsulURL() + "/v1/txn"
2731

2832
// build our request
29-
i.logger.PrintDebug("CONSUL: creating PUT request")
33+
i.logger.PrintDebug("CONSUL: creating PUT request for Batch " + batch)
3034
req, err := http.NewRequest("PUT", consulUrl, bytes.NewBuffer(jsonPayload))
3135
if err != nil {
32-
util.ExitError(errors.New("NewRequestPUT: "+err.Error()), util.ErrorFailedConsulConnection, i.logger)
36+
util.ExitError(errors.New("NewRequestPUT"+err.Error()+" in Batch "+batch), util.ErrorFailedConsulConnection, i.logger)
3337
}
3438

3539
// Set ACL token
@@ -40,10 +44,10 @@ func (i *importer) processConsulTransaction(transactions []entities.ConsulTxn) {
4044
// Send the request via a client
4145
// Do sends an HTTP request and
4246
// returns an HTTP response
43-
i.logger.PrintDebug("CONSUL: calling PUT request")
47+
i.logger.PrintDebug("CONSUL: calling PUT request for Batch " + batch)
4448
resp, err := i.client.Do(req)
4549
if err != nil {
46-
util.ExitError(errors.New("DoPUT: "+err.Error()), util.ErrorFailedConsulConnection, i.logger)
50+
util.ExitError(errors.New("DoPUT: "+err.Error()+" for Batch "+batch), util.ErrorFailedConsulConnection, i.logger)
4751
}
4852

4953
// Clean response after function ends
@@ -54,21 +58,21 @@ func (i *importer) processConsulTransaction(transactions []entities.ConsulTxn) {
5458
}()
5559

5660
// Read the response body
57-
i.logger.PrintDebug("CONSUL: reading PUT response")
61+
i.logger.PrintDebug("CONSUL: reading PUT response from Batch " + batch)
5862
bodyBytes, err := ioutil.ReadAll(resp.Body)
5963
if err != nil {
60-
util.ExitError(errors.New("ReadPutResponse: "+err.Error()), util.ErrorFailedReadingResponse, i.logger)
64+
util.ExitError(errors.New("ReadPutResponse: "+err.Error()+" in Batch "+batch), util.ErrorFailedReadingResponse, i.logger)
6165
}
6266

6367
// Cast response to string
6468
bodyString := string(bodyBytes)
6569

6670
if resp.StatusCode != 200 {
67-
util.ExitError(errors.New("TransactionError: "+bodyString), util.ErrorFailedConsulTxn, i.logger)
71+
util.ExitError(errors.New("TransactionError: "+bodyString+" in Batch "+batch), util.ErrorFailedConsulTxn, i.logger)
6872
}
6973

7074
// All good. Output some status for each transaction operation
7175
for _, txn := range transactions {
72-
i.logger.PrintInfo("Operation: " + *txn.KV.Verb + " Path: " + *txn.KV.Key)
76+
i.logger.PrintInfo("Operation: " + *txn.KV.Verb + " Path: " + *txn.KV.Key + " Batch: " + batch)
7377
}
7478
}

‎internal/importer/helper.go‎

Lines changed: 55 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,10 @@
11
package importer
22

33
import (
4+
"strings"
5+
46
"github.com/miniclip/gonsul/internal/entities"
57
"github.com/miniclip/gonsul/internal/util"
6-
"strings"
78

89
"github.com/cbroglie/mustache"
910
"github.com/olekukonko/tablewriter"
@@ -16,6 +17,7 @@ import (
1617
"net/http"
1718
"os"
1819
"path"
20+
"strconv"
1921
)
2022

2123
// createOperationMatrix ...
@@ -146,17 +148,56 @@ func (i *importer) printOperations(matrix entities.OperationMatrix, printWhat st
146148
if matrix.GetTotalOps() > 0 {
147149
// Instantiate our table and set table header
148150
table := tablewriter.NewWriter(os.Stdout)
149-
table.SetHeader([]string{"", "OPERATION NAME", "CONSUL VERB", "PATH"})
151+
table.SetHeader([]string{"", "BATCH", "OP INDEX", "OPERATION NAME", "CONSUL VERB", "PATH"})
150152
// Align our rows
151153
table.SetAlignment(tablewriter.ALIGN_LEFT)
154+
155+
// Initialize the batch counter
156+
batch := 1
157+
opIndex := 0
158+
159+
var transactions []entities.ConsulTxn
160+
var newTransactions []entities.ConsulTxn
161+
152162
// Loop each operation and add to table
153163
for _, op := range matrix.GetOperations() {
164+
154165
if printWhat == entities.OperationAll || printWhat == op.GetType() {
166+
var TxnKV entities.ConsulTxnKV
167+
var warning string
168+
169+
// generate the actual payload to calculate it's lenght
170+
verb := op.GetVerb()
171+
path := op.GetPath()
155172
if op.GetType() == entities.OperationDelete {
156-
table.Append([]string{"!!", op.GetType(), op.GetVerb(), op.GetPath()})
173+
warning = "!!"
174+
TxnKV = entities.ConsulTxnKV{Verb: &verb, Key: &path}
157175
} else {
158-
table.Append([]string{"", op.GetType(), op.GetVerb(), op.GetPath()})
176+
warning = ""
177+
val := op.GetValue()
178+
TxnKV = entities.ConsulTxnKV{Verb: &verb, Key: &path, Value: &val}
179+
}
180+
181+
// add the next transaction and check payload lenght
182+
newTransactions = transactions
183+
newTransactions = append(newTransactions, entities.ConsulTxn{KV: TxnKV})
184+
newPayloadSize := i.getTransactionsPayloadSize(&newTransactions)
185+
186+
// If the next transaction brings us over the maximum payload size,
187+
// or the maximum transaction per batch limit is reached, start a new batch
188+
if newPayloadSize > maximumPayloadSize || len(transactions) == consulTxnLimit {
189+
// reset transactions and add the next transaction
190+
transactions = []entities.ConsulTxn{}
191+
// start a new batch counter
192+
opIndex = 0
193+
batch++
159194
}
195+
196+
transactions = append(transactions, entities.ConsulTxn{KV: TxnKV})
197+
198+
table.Append([]string{warning, strconv.Itoa(batch), strconv.Itoa(opIndex), op.GetType(), op.GetVerb(), op.GetPath()})
199+
200+
opIndex++
160201
}
161202
}
162203
// Outputs ASCII table
@@ -178,3 +219,13 @@ func (i *importer) setDeletesToLogger(matrix entities.OperationMatrix) {
178219
}
179220
}
180221
}
222+
223+
// Get the payload size for a slice of transactions
224+
func (i *importer) getTransactionsPayloadSize(transactions *[]entities.ConsulTxn) int {
225+
payload, err := json.Marshal(&transactions)
226+
if err != nil {
227+
util.ExitError(errors.New("Marshal: "+err.Error()), util.ErrorFailedJsonEncode, i.logger)
228+
}
229+
230+
return len(string(payload))
231+
}

‎internal/importer/importer.go‎

Lines changed: 16 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,6 @@ import (
55
"github.com/miniclip/gonsul/internal/entities"
66
"github.com/miniclip/gonsul/internal/util"
77

8-
"encoding/json"
98
"errors"
109
"fmt"
1110
"net/http"
@@ -41,11 +40,11 @@ func (i *importer) Start(localData map[string]string) {
4140
// Create our operations Matrix
4241
ops = i.createOperationMatrix(liveData, localData)
4342

43+
// Print operation table
44+
i.printOperations(ops, entities.OperationAll)
4445
// Check if it's a dry run
4546
if i.config.GetStrategy() == config.StrategyDry {
46-
// Print matrix and exit
47-
i.printOperations(ops, entities.OperationAll)
48-
47+
// Exit after having printed the operations table
4948
return
5049
}
5150

@@ -72,7 +71,11 @@ func (i *importer) processOperations(matrix entities.OperationMatrix) {
7271
util.ExitError(errors.New(""), util.ErrorDeleteNotAllowed, i.logger)
7372
}
7473

74+
// Initialize the batch counter
75+
batch := 1
76+
7577
var transactions []entities.ConsulTxn
78+
var newTransactions []entities.ConsulTxn
7679

7780
// Fill our channel to indicate a non interruptible work (It stops here if interruption in progress)
7881
i.config.WorkingChan() <- true
@@ -84,12 +87,6 @@ func (i *importer) processOperations(matrix entities.OperationMatrix) {
8487
verb := op.GetVerb()
8588
path := op.GetPath()
8689

87-
currentPayload, err := json.Marshal(transactions)
88-
if err != nil {
89-
util.ExitError(errors.New("Marshal: "+err.Error()), util.ErrorFailedJsonEncode, i.logger)
90-
}
91-
currentPayloadSize := len(currentPayload)
92-
9390
var TxnKV entities.ConsulTxnKV
9491

9592
if op.GetType() == entities.OperationDelete {
@@ -99,32 +96,24 @@ func (i *importer) processOperations(matrix entities.OperationMatrix) {
9996
TxnKV = entities.ConsulTxnKV{Verb: &verb, Key: &path, Value: &val}
10097
}
10198

102-
// If the next transaction brings us over the maximum payload size, flush the current transactions
103-
nextOpPayload, err := json.Marshal(TxnKV)
104-
if err != nil {
105-
util.ExitError(errors.New("Marshal: "+err.Error()), util.ErrorFailedJsonEncode, i.logger)
106-
}
99+
// add the next transaction and check payload lenght
100+
newTransactions = transactions
101+
newTransactions = append(transactions, entities.ConsulTxn{KV: TxnKV})
102+
newPayloadSize := i.getTransactionsPayloadSize(&newTransactions)
107103

108-
if currentPayloadSize+len(nextOpPayload) > maximumPayloadSize {
109-
i.processConsulTransaction(transactions)
104+
if newPayloadSize > maximumPayloadSize || len(transactions) == consulTxnLimit {
105+
i.processConsulTransaction(transactions, batch)
110106
transactions = []entities.ConsulTxn{}
107+
108+
batch++
111109
}
112110

113111
transactions = append(transactions, entities.ConsulTxn{KV: TxnKV})
114-
115-
if len(transactions) == consulTxnLimit {
116-
// Flush current transactions because we hit max operation per transaction
117-
// One day Consul will release an API endpoint from where we can get this limit
118-
// so we do can stop hardcoding this constant
119-
i.processConsulTransaction(transactions)
120-
// Reset our transaction array
121-
transactions = []entities.ConsulTxn{}
122-
}
123112
}
124113

125114
// Do we have transactions to process
126115
if len(transactions) > 0 {
127-
i.processConsulTransaction(transactions)
116+
i.processConsulTransaction(transactions, batch)
128117
}
129118

130119
// Consume our channel, to re-allow application interruption

0 commit comments

Comments
 (0)