Added async behavior.

pkg/kubelet/kubelet.go

@@ -639,55 +639,77 @@ func (kl *Kubelet) createNetworkContainer(manifest *api.ContainerManifest) (Dock
 	return kl.runContainer(manifest, container, "")
 }
 
-// Sync the configured list of containers (desired state) with the host current state
+func (kl *Kubelet) syncManifest(manifest *api.ContainerManifest, dockerIdsToKeep *map[DockerId]bool, mapLock *sync.Mutex) error {
-func (kl *Kubelet) SyncManifests(config []api.ContainerManifest) error {
-	glog.Infof("Desired: %+v", config)
-	var err error
-	dockerIdsToKeep := map[DockerId]bool{}
-
-	// Check for any containers that need starting
-	for _, manifest := range config {
 	// Make sure we have a network container
-		netId, err := kl.getNetworkContainerId(&manifest)
+	netId, err := kl.getNetworkContainerId(manifest)
 	if err != nil {
-			glog.Errorf("Failed to introspect network container. (%v)  Skipping container %s", err, manifest.ID)
+		glog.Errorf("Failed to introspect network container. (%v)  Skipping container %s", err, manifest.Id)
-			continue
+		return err
 	}
 	if netId == "" {
 		glog.Infof("Network container doesn't exist, creating")
-			netId, err = kl.createNetworkContainer(&manifest)
+		netId, err = kl.createNetworkContainer(manifest)
 		if err != nil {
-				glog.Errorf("Failed to create network container: %v Skipping container %s", err, manifest.ID)
+			glog.Errorf("Failed to introspect network container. (%v)  Skipping container %s", err, manifest.ID)
-				continue
+			return err
 		}
 	}
-		dockerIdsToKeep[netId] = true
+	{
-
+		mapLock.Lock()
+		defer mapLock.Unlock()
+		(*dockerIdsToKeep)[netId] = true
+	}
 	for _, container := range manifest.Containers {
-			containerId, err := kl.getContainerId(&manifest, &container)
+		containerId, err := kl.getContainerId(manifest, &container)
 		if err != nil {
-				glog.Errorf("Error detecting container: %v skipping.", err)
+			glog.Errorf("Error finding container: %v skipping.", err)
 			continue
 		}
 		if containerId == "" {
 			glog.Infof("%+v doesn't exist, creating", container)
 			kl.DockerPuller.Pull(container.Image)
 			if err != nil {
-					glog.Errorf("Error pulling container: %v", err)
+				glog.Errorf("Failed to create network container: %v Skipping container %s", err, manifest.ID)
 				continue
 			}
-				containerId, err = kl.runContainer(&manifest, &container, "container:"+string(netId))
+			containerId, err = kl.runContainer(manifest, &container, "container:"+string(netId))
 			if err != nil {
 				// TODO(bburns) : Perhaps blacklist a container after N failures?
-					glog.Errorf("Error creating container: %v", err)
+				glog.Errorf("Error running container: %v skipping.", err)
 				continue
 			}
 		} else {
 			glog.V(1).Infof("%s exists as %v", container.Name, containerId)
 		}
-			dockerIdsToKeep[containerId] = true
+		{
+			mapLock.Lock()
+			defer mapLock.Unlock()
+			(*dockerIdsToKeep)[containerId] = true
 		}
 	}
+	return nil
+}
+
+// Sync the configured list of containers (desired state) with the host current state
+func (kl *Kubelet) SyncManifests(config []api.ContainerManifest) error {
+	glog.Infof("Desired: %+v", config)
+	var err error
+	dockerIdsToKeep := map[DockerId]bool{}
+	mapLock := sync.Mutex{}
+	waitGroup := sync.WaitGroup{}
+
+	// Check for any containers that need starting
+	for _, manifest := range config {
+		waitGroup.Add(1)
+		go func() {
+			err := kl.syncManifest(&manifest, &dockerIdsToKeep, &mapLock)
+			if err != nil {
+				glog.Errorf("Error syncing manifest: %v skipping.", err)
+			}
+			waitGroup.Done()
+		}()
+	}
+	waitGroup.Wait()
 
 	// Kill any containers we don't need
 	existingContainers, err := kl.getDockerContainers()