@@ -34,7 +34,7 @@ type Indexer interface {
3434 Index (rt * runtime.Runtime , rebuild , cleanup bool ) (string , error )
3535 Stats () Stats
3636
37- GetESLastModified (index string ) (time.Time , error )
37+ GetESLastModified (ctx context. Context , index string ) (time.Time , error )
3838 GetDBLastModified (ctx context.Context , db * sql.DB ) (time.Time , error )
3939}
4040
@@ -98,9 +98,9 @@ func (i *baseIndexer) recordActivity(indexed, deleted int, elapsed time.Duration
9898type infoResponse map [string ]interface {}
9999
100100// FindIndexes finds all our physical indexes
101- func (i * baseIndexer ) FindIndexes () []string {
101+ func (i * baseIndexer ) FindIndexes (ctx context. Context ) []string {
102102 response := infoResponse {}
103- _ , err := utils .MakeJSONRequest (http .MethodGet , fmt .Sprintf ("%s/%s" , i .elasticURL , i .name ), nil , & response )
103+ _ , err := utils .MakeJSONRequest (ctx , http .MethodGet , fmt .Sprintf ("%s/%s" , i .elasticURL , i .name ), nil , & response )
104104 indexes := make ([]string , 0 )
105105
106106 // error could mean a variety of things, but we'll figure that out later
@@ -128,7 +128,7 @@ func (i *baseIndexer) FindIndexes() []string {
128128// that index to `contacts`.
129129//
130130// If the day-specific name already exists, we append a .1 or .2 to the name.
131- func (i * baseIndexer ) createNewIndex (def * IndexDefinition ) (string , error ) {
131+ func (i * baseIndexer ) createNewIndex (ctx context. Context , def * IndexDefinition ) (string , error ) {
132132 // create our day-specific name
133133 index := fmt .Sprintf ("%s_%s" , i .name , time .Now ().Format ("2006_01_02" ))
134134 idx := 0
@@ -152,7 +152,7 @@ func (i *baseIndexer) createNewIndex(def *IndexDefinition) (string, error) {
152152 // create the new index
153153 settings := jsonx .MustMarshal (def )
154154
155- _ , err := utils .MakeJSONRequest (http .MethodPut , fmt .Sprintf ("%s/%s" , i .elasticURL , index ), settings , nil )
155+ _ , err := utils .MakeJSONRequest (ctx , http .MethodPut , fmt .Sprintf ("%s/%s" , i .elasticURL , index ), settings , nil )
156156 if err != nil {
157157 return "" , err
158158 }
@@ -185,11 +185,11 @@ type removeAliasCommand struct {
185185}
186186
187187// maps this indexer's alias to the new physical index, removing existing aliases if they exist
188- func (i * baseIndexer ) updateAlias (newIndex string ) error {
188+ func (i * baseIndexer ) updateAlias (ctx context. Context , newIndex string ) error {
189189 commands := make ([]interface {}, 0 )
190190
191191 // find existing physical indexes
192- existing := i .FindIndexes ()
192+ existing := i .FindIndexes (ctx )
193193 for _ , idx := range existing {
194194 remove := removeAliasCommand {}
195195 remove .Remove .Alias = i .name
@@ -207,7 +207,7 @@ func (i *baseIndexer) updateAlias(newIndex string) error {
207207
208208 aliasJSON := jsonx .MustMarshal (aliasCommand {Actions : commands })
209209
210- _ , err := utils .MakeJSONRequest (http .MethodPost , fmt .Sprintf ("%s/_aliases" , i .elasticURL ), aliasJSON , nil )
210+ _ , err := utils .MakeJSONRequest (ctx , http .MethodPost , fmt .Sprintf ("%s/_aliases" , i .elasticURL ), aliasJSON , nil )
211211
212212 i .log ().Info ("updated alias" , "index" , newIndex )
213213
@@ -222,9 +222,9 @@ type healthResponse struct {
222222}
223223
224224// removes all indexes that are older than the currently active index
225- func (i * baseIndexer ) cleanupIndexes () error {
225+ func (i * baseIndexer ) cleanupIndexes (ctx context. Context ) error {
226226 // find our current indexes
227- currents := i .FindIndexes ()
227+ currents := i .FindIndexes (ctx )
228228
229229 // no current indexes? this a noop
230230 if len (currents ) == 0 {
@@ -233,7 +233,7 @@ func (i *baseIndexer) cleanupIndexes() error {
233233
234234 // find all the current indexes
235235 healthResponse := healthResponse {}
236- _ , err := utils .MakeJSONRequest (http .MethodGet , fmt .Sprintf ("%s/%s" , i .elasticURL , "_cluster/health?level=indices" ), nil , & healthResponse )
236+ _ , err := utils .MakeJSONRequest (ctx , http .MethodGet , fmt .Sprintf ("%s/%s" , i .elasticURL , "_cluster/health?level=indices" ), nil , & healthResponse )
237237 if err != nil {
238238 return err
239239 }
@@ -242,7 +242,7 @@ func (i *baseIndexer) cleanupIndexes() error {
242242 for key := range healthResponse .Indices {
243243 if strings .HasPrefix (key , i .name ) && strings .Compare (key , currents [0 ]) < 0 {
244244 slog .Info ("removing old index" , "index" , key )
245- _ , err = utils .MakeJSONRequest (http .MethodDelete , fmt .Sprintf ("%s/%s" , i .elasticURL , key ), nil , nil )
245+ _ , err = utils .MakeJSONRequest (ctx , http .MethodDelete , fmt .Sprintf ("%s/%s" , i .elasticURL , key ), nil , nil )
246246 if err != nil {
247247 return err
248248 }
@@ -268,11 +268,11 @@ type indexResponse struct {
268268}
269269
270270// indexes the batch of contacts
271- func (i * baseIndexer ) indexBatch (index string , batch []byte ) (int , int , int , error ) {
271+ func (i * baseIndexer ) indexBatch (ctx context. Context , index string , batch []byte ) (int , int , int , error ) {
272272 response := indexResponse {}
273273 indexURL := fmt .Sprintf ("%s/%s/_bulk" , i .elasticURL , index )
274274
275- _ , err := utils .MakeJSONRequest (http .MethodPut , indexURL , batch , & response )
275+ _ , err := utils .MakeJSONRequest (ctx , http .MethodPut , indexURL , batch , & response )
276276 if err != nil {
277277 return 0 , 0 , 0 , err
278278 }
@@ -324,12 +324,13 @@ type queryResponse struct {
324324}
325325
326326// GetESLastModified queries a concrete index and finds the last modified document, returning its modified time
327- func (i * baseIndexer ) GetESLastModified (index string ) (time.Time , error ) {
327+ func (i * baseIndexer ) GetESLastModified (ctx context. Context , index string ) (time.Time , error ) {
328328 lastModified := time.Time {}
329329
330330 // get the newest document on our index
331331 queryResponse := & queryResponse {}
332332 _ , err := utils .MakeJSONRequest (
333+ ctx ,
333334 http .MethodPost ,
334335 fmt .Sprintf ("%s/%s/_search" , i .elasticURL , index ),
335336 []byte (`{ "sort": [{ "modified_on_mu": "desc" }], "_source": {"includes": ["modified_on", "id"]}, "size": 1, "track_total_hits": false}` ),
0 commit comments