-
Notifications
You must be signed in to change notification settings - Fork 782
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
6 changed files
with
240 additions
and
62 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,143 @@ | ||
| package dependency | ||
|
|
||
| import ( | ||
| "log" | ||
| "net/url" | ||
| "slices" | ||
| "time" | ||
|
|
||
| capi "github.com/hashicorp/consul/api" | ||
| "github.com/pkg/errors" | ||
| ) | ||
|
|
||
| const ( | ||
| exportedServicesEndpointLabel = "list.exportedServices" | ||
|
|
||
| // CatalogExportedServicesQuerySleepTime is the amount of time to sleep between | ||
| // queries, since the endpoint does not support blocking queries. | ||
| CatalogExportedServicesQuerySleepTime = 15 * time.Second | ||
| ) | ||
|
|
||
| // Ensure implements | ||
| var _ Dependency = (*CatalogExportedServicesQuery)(nil) | ||
|
|
||
| // CatalogExportedServicesQuery is the representation of a requested catalog partitions | ||
| // dependency from inside a template. | ||
| type CatalogExportedServicesQuery struct { | ||
| stopCh chan struct{} | ||
| partition string | ||
| } | ||
|
|
||
| type ExportedService struct { | ||
| // Name of the service | ||
| Service string | ||
|
|
||
| // Partition of the service | ||
| Partition string | ||
|
|
||
| // Namespace of the service | ||
| Namespace string | ||
|
|
||
| // Consumers is a list of downstream consumers of the service. | ||
| Consumers ResolvedConsumers | ||
| } | ||
|
|
||
| type ResolvedConsumers struct { | ||
| Peers []string | ||
| Partitions []string | ||
| } | ||
|
|
||
| func fromConsulExportedService(svc capi.ResolvedExportedService) ExportedService { | ||
| return ExportedService{ | ||
| Service: svc.Service, | ||
| Consumers: ResolvedConsumers{ | ||
| Peers: slices.Clone(svc.Consumers.Peers), | ||
| Partitions: slices.Clone(svc.Consumers.Partitions), | ||
| }, | ||
| } | ||
| } | ||
|
|
||
| // NewCatalogExportedServicesQuery parses a string of the format @dc. | ||
| func NewCatalogExportedServicesQuery(s string) (*CatalogExportedServicesQuery, error) { | ||
| return &CatalogExportedServicesQuery{ | ||
| stopCh: make(chan struct{}, 1), | ||
| partition: s, | ||
| }, nil | ||
| } | ||
|
|
||
| func (c *CatalogExportedServicesQuery) Fetch(clients *ClientSet, opts *QueryOptions) (interface{}, *ResponseMetadata, error) { | ||
| opts = opts.Merge(&QueryOptions{ | ||
| ConsulPartition: c.partition, | ||
| }) | ||
|
|
||
| log.Printf("[TRACE] %s: GET %s", c, &url.URL{ | ||
| Path: "/v1/exported-services", | ||
| RawQuery: opts.String(), | ||
| }) | ||
|
|
||
| // This is certainly not elegant, but the partitions endpoint does not support | ||
| // blocking queries, so we are going to "fake it until we make it". When we | ||
| // first query, the LastIndex will be "0", meaning we should immediately | ||
| // return data, but future calls will include a LastIndex. If we have a | ||
| // LastIndex in the query metadata, sleep for 15 seconds before asking Consul | ||
| // again. | ||
| // | ||
| // This is probably okay given the frequency in which partitions actually | ||
| // change, but is technically not edge-triggering. | ||
| if opts.WaitIndex != 0 { | ||
| log.Printf("[TRACE] %s: long polling for %s", c, CatalogDatacentersQuerySleepTime) | ||
|
|
||
| select { | ||
| case <-c.stopCh: | ||
| return nil, nil, ErrStopped | ||
| case <-time.After(CatalogExportedServicesQuerySleepTime): | ||
| } | ||
| } | ||
|
|
||
| // TODO Consider using a proper context | ||
| consulExportedServices, qm, err := clients.Consul().ExportedServices(opts.ToConsulOpts()) | ||
| if err != nil { | ||
| return nil, nil, errors.Wrapf(err, c.String()) | ||
| } | ||
|
|
||
| exportedServices := make([]ExportedService, 0, len(consulExportedServices)) | ||
| for _, svc := range consulExportedServices { | ||
| exportedServices = append(exportedServices, fromConsulExportedService(svc)) | ||
| } | ||
|
|
||
| log.Printf("[TRACE] %s: returned %d results", c, len(exportedServices)) | ||
|
|
||
| slices.SortStableFunc(exportedServices, func(i, j ExportedService) int { | ||
| if i.Service < j.Service { | ||
| return -1 | ||
| } else if i.Service > j.Service { | ||
| return 1 | ||
| } | ||
| return 0 | ||
| }) | ||
|
|
||
| rm := &ResponseMetadata{ | ||
| LastContact: qm.LastContact, | ||
| LastIndex: qm.LastIndex, | ||
| } | ||
|
|
||
| return exportedServices, rm, nil | ||
| } | ||
|
|
||
| // CanShare returns if this dependency is shareable. | ||
| // TODO What is this? | ||
| func (c *CatalogExportedServicesQuery) CanShare() bool { | ||
| return true | ||
| } | ||
|
|
||
| func (c *CatalogExportedServicesQuery) String() string { | ||
| return exportedServicesEndpointLabel | ||
| } | ||
|
|
||
| func (c *CatalogExportedServicesQuery) Stop() { | ||
| close(c.stopCh) | ||
| } | ||
|
|
||
| func (c *CatalogExportedServicesQuery) Type() Type { | ||
| return TypeConsul | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.