Add support for polling a Kubernetes Master server #91
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
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
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,179 @@ | ||
| package main | ||
|
|
||
| import ( | ||
| "flag" | ||
| "log" | ||
| "net" | ||
| "sync" | ||
| "time" | ||
|
|
||
| "encoding/json" | ||
|
|
||
| "github.com/GoogleCloudPlatform/kubernetes/pkg/api" | ||
| "github.com/GoogleCloudPlatform/kubernetes/pkg/client" | ||
| pconfig "github.com/GoogleCloudPlatform/kubernetes/pkg/proxy/config" | ||
| "github.com/GoogleCloudPlatform/kubernetes/pkg/util" | ||
| "github.com/coreos/go-etcd/etcd" | ||
| "github.com/skynetservices/skydns/msg" | ||
| ) | ||
|
|
||
| // The periodic interval for checking the state of things. | ||
| const syncInterval = 5 * time.Second | ||
|
There was a problem hiding this comment. There is no way to get callbacks as of yet, I presume? |
||
|
|
||
| type KubernetesSync struct { | ||
| mu sync.Mutex // protects serviceMap | ||
| serviceMap map[string]*serviceInfo | ||
| eclient *etcd.Client | ||
| } | ||
|
|
||
| func NewKubernetesSync(client *etcd.Client) *KubernetesSync { | ||
| ks := &KubernetesSync{ | ||
| serviceMap: make(map[string]*serviceInfo), | ||
| eclient: client, | ||
| } | ||
| return ks | ||
| } | ||
|
|
||
| // SyncLoop runs periodic work. This is expected to run as a goroutine or as the main loop of the app. It does not return. | ||
|
|
||
| func (ksync *KubernetesSync) SyncLoop() { | ||
| for { | ||
| select { | ||
| case <-time.After(syncInterval): | ||
| log.Println("Periodic sync") | ||
| ksync.ensureDNS() | ||
| } | ||
| } | ||
| } | ||
|
|
||
| // Ensure that dns records exist for all services. | ||
|
There was a problem hiding this comment. I'm confused by the first line of this comment and then again by the second line? If not sure, remove it and see what breaks? |
||
| // This seems a bit redundant. TBD - remove? | ||
| func (ksync *KubernetesSync) ensureDNS() { | ||
| ksync.mu.Lock() | ||
| defer ksync.mu.Unlock() | ||
| for name, info := range ksync.serviceMap { | ||
| err := ksync.addDNS(name, info) | ||
| if err != nil { | ||
| log.Println("Failed to ensure dns for %q: %s", name, err) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| // OnUpdate manages the active set of service records. | ||
| // Active service records get ttl bumps if found in the update set or | ||
| // removed if missing from the update set. | ||
|
|
||
|
|
||
| func (ksync *KubernetesSync) OnUpdate(services []api.Service) { | ||
| activeServices := util.StringSet{} | ||
|
There was a problem hiding this comment. util.StringSet? Is the some uber cool thing better than []string? There was a problem hiding this comment. For the Kubernetes team, yes :) Some of this code is modeled directly from their internal polling routines. |
||
| for _, service := range services { | ||
| activeServices.Insert(service.ID) | ||
| info, exists := ksync.getServiceInfo(service.ID) | ||
| serviceIP := net.ParseIP(service.PortalIP) | ||
| if exists && (info.portalPort != service.Port || !info.portalIP.Equal(serviceIP)) { | ||
| err := ksync.removeDNS(service.ID, info) | ||
| if err != nil { | ||
| log.Printf("Failed to remove dns for %q: %s\n", service.ID, err) | ||
| } | ||
| } | ||
| log.Printf("Adding new service %q at %s:%d/%s (local :%d)\n", service.ID, serviceIP, service.Port, service.Protocol, service.ProxyPort) | ||
| si := &serviceInfo{ | ||
| proxyPort: service.ProxyPort, | ||
| protocol: service.Protocol, | ||
| active: true, | ||
| } | ||
| ksync.setServiceInfo(service.ID, si) | ||
| si.portalIP = serviceIP | ||
| si.portalPort = service.Port | ||
| err := ksync.addDNS(service.ID, si) | ||
| if err != nil { | ||
| log.Println("Failed to add dns %q: %s", service.ID, err) | ||
| } | ||
| } | ||
| ksync.mu.Lock() | ||
| defer ksync.mu.Unlock() | ||
| for name, info := range ksync.serviceMap { | ||
| if !activeServices.Has(name) { | ||
| err := ksync.removeDNS(name, info) | ||
| if err != nil { | ||
| log.Println("Failed to remove dns for %q: %s", name, err) | ||
| } | ||
| delete(ksync.serviceMap, name) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| func (ksync *KubernetesSync) getServiceInfo(service string) (*serviceInfo, bool) { | ||
| ksync.mu.Lock() | ||
| defer ksync.mu.Unlock() | ||
| info, ok := ksync.serviceMap[service] | ||
| return info, ok | ||
| } | ||
|
|
||
| func (ksync *KubernetesSync) setServiceInfo(service string, info *serviceInfo) { | ||
| ksync.mu.Lock() | ||
| defer ksync.mu.Unlock() | ||
| ksync.serviceMap[service] = info | ||
| } | ||
|
|
||
| func (ksync *KubernetesSync) removeDNS(service string, info *serviceInfo) error { | ||
| record := service + "." + config.Domain | ||
| // Remove from SkyDNS registration | ||
| log.Printf("removing %s from DNS", record) | ||
| _, err := ksync.eclient.Delete(msg.Path(record), true) | ||
| return err | ||
| } | ||
|
|
||
| func (ksync *KubernetesSync) addDNS(service string, info *serviceInfo) error { | ||
| // ADD to SkyDNS registry | ||
| svc := msg.Service{ | ||
| Host: info.portalIP.String(), | ||
| Port: info.portalPort, | ||
| Priority: 10, | ||
| Weight: 10, | ||
| Ttl: 30, | ||
| } | ||
| b, err := json.Marshal(svc) | ||
| record := service + "." + config.Domain | ||
| //Set with no TTL, and hope that kubernetes events are accurate. | ||
| //TODO(BJK) Think this through a little more | ||
|
|
||
| log.Printf("Setting dns record: %v\n", record) | ||
|
There was a problem hiding this comment. Please use smaller case 'S' more inline with the rest of the logging. |
||
| _, err = ksync.eclient.Set(msg.Path(record), string(b), uint64(0)) | ||
| return err | ||
| } | ||
|
|
||
| type serviceInfo struct { | ||
| portalIP net.IP | ||
| portalPort int | ||
| protocol api.Protocol | ||
| proxyPort int | ||
| mu sync.Mutex // protects active | ||
| active bool | ||
| } | ||
|
|
||
| func init() { | ||
| client.BindClientConfigFlags(flag.CommandLine, clientConfig) | ||
| } | ||
|
|
||
| func WatchKubernetes(eclient *etcd.Client) { | ||
| serviceConfig := pconfig.NewServiceConfig() | ||
| endpointsConfig := pconfig.NewEndpointsConfig() | ||
|
|
||
| // define api config source | ||
| if clientConfig.Host != "" { | ||
| log.Println("Using api calls to get Kubernetes config %v", clientConfig.Host) | ||
| client, err := client.New(clientConfig) | ||
| if err != nil { | ||
| log.Fatalf("Kubernetes requested, but received invalid API configuration: %v", err) | ||
| } | ||
| pconfig.NewSourceAPI( | ||
| client, | ||
| 30*time.Second, | ||
| serviceConfig.Channel("api"), | ||
| endpointsConfig.Channel("api"), | ||
| ) | ||
| } | ||
| ks := NewKubernetesSync(eclient) | ||
| // Wire skydns to handle changes to services | ||
| serviceConfig.RegisterHandler(ks) | ||
| ks.SyncLoop() | ||
| } | ||
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
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
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
why primitive
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Because their service definitions are evolving, eventually they will have more things published other than services, we'll support those as they come.