// handleMaster performs one loop of master locking.
// on success it returns <master>, nil
// on error it returns "", err
// in situations where you should try again due to concurrent state changes (e.g. another actor simultaneously acquiring the lock)
// it returns "", nil
func (e *etcdMasterElector) handleMaster(path, id string, ttl uint64) (string, error) {
	res, err := e.etcd.Get(path, false, false)

	// Unexpected error, bail out
	if err != nil && !tools.IsEtcdNotFound(err) {
		return "", err
	}

	// There is no master, try to become the master.
	if err != nil && tools.IsEtcdNotFound(err) {
		return e.becomeMaster(path, id, ttl)
	}

	// This should never happen.
	if res.Node == nil {
		return "", fmt.Errorf("unexpected response: %#v", res)
	}

	// We're not the master, just return the current value
	if res.Node.Value != id {
		return res.Node.Value, nil
	}

	// We are the master, try to extend out lease
	return e.extendMaster(path, id, ttl, res)
}

// DeletePod deletes an existing pod specified by its ID.
func (registry *EtcdRegistry) DeletePod(podID string) error {
	var pod api.Pod
	podKey := makePodKey(podID)
	err := registry.helper.ExtractObj(podKey, &pod, false)
	if tools.IsEtcdNotFound(err) {
		return apiserver.NewNotFoundErr("pod", podID)
	}
	if err != nil {
		return err
	}

	// First delete the pod, so a scheduler doesn't notice it getting removed from the
	// machine and attempt to put it somewhere.
	err = registry.helper.Delete(podKey, true)
	if tools.IsEtcdNotFound(err) {
		return apiserver.NewNotFoundErr("pod", podID)
	}
	if err != nil {
		return err
	}

	machine := pod.DesiredState.Host
	if machine == "" {
		// Pod was never scheduled anywhere, just return.
		return nil
	}

	// Next, remove the pod from the machine atomically.
	contKey := makeContainerKey(machine)
	return registry.helper.AtomicUpdate(contKey, &api.ContainerManifestList{}, func(in interface{}) (interface{}, error) {
		manifests := in.(*api.ContainerManifestList)
		newManifests := make([]api.ContainerManifest, 0, len(manifests.Items))
		found := false
		for _, manifest := range manifests.Items {
			if manifest.ID != podID {
				newManifests = append(newManifests, manifest)
			} else {
				found = true
			}
		}
		if !found {
			// This really shouldn't happen, it indicates something is broken, and likely
			// there is a lost pod somewhere.
			// However it is "deleted" so log it and move on
			glog.Infof("Couldn't find: %s in %#v", podID, manifests)
		}
		manifests.Items = newManifests
		return manifests, nil
	})
}

// DeleteService deletes a Service specified by its name.
func (r *Registry) DeleteService(name string) error {
	key := makeServiceKey(name)
	err := r.Delete(key, true)
	if tools.IsEtcdNotFound(err) {
		return errors.NewNotFound("service", name)
	}
	if err != nil {
		return err
	}
	key = makeServiceEndpointsKey(name)
	err = r.Delete(key, true)
	if !tools.IsEtcdNotFound(err) {
		return err
	}
	return nil
}

func (registry *EtcdRegistry) deletePodFromMachine(machine, podID string) error {
	// First delete the pod, so a scheduler doesn't notice it getting removed from the
	// machine and attempt to put it somewhere.
	podKey := makePodKey(machine, podID)
	_, err := registry.etcdClient.Delete(podKey, true)
	if tools.IsEtcdNotFound(err) {
		return apiserver.NewNotFoundErr("pod", podID)
	}
	if err != nil {
		return err
	}

	// Next, remove the pod from the machine atomically.
	contKey := makeContainerKey(machine)
	return registry.helper().AtomicUpdate(contKey, &[]api.ContainerManifest{}, func(in interface{}) (interface{}, error) {
		manifests := *in.(*[]api.ContainerManifest)
		newManifests := make([]api.ContainerManifest, 0, len(manifests))
		found := false
		for _, manifest := range manifests {
			if manifest.ID != podID {
				newManifests = append(newManifests, manifest)
			} else {
				found = true
			}
		}
		if !found {
			// This really shouldn't happen, it indicates something is broken, and likely
			// there is a lost pod somewhere.
			// However it is "deleted" so log it and move on
			glog.Infof("Couldn't find: %s in %#v", podID, manifests)
		}
		return newManifests, nil
	})
}

// GetAndTestEtcdClient creates an etcd client based on the provided config. It will attempt to
// connect to the etcd server and block until the server responds at least once, or return an
// error if the server never responded.
func GetAndTestEtcdClient(etcdClientInfo configapi.EtcdConnectionInfo) (*etcdclient.Client, error) {
	// etcd does a poor job of setting up the transport - use the Kube client stack
	transport, err := client.TransportFor(&client.Config{
		TLSClientConfig: client.TLSClientConfig{
			CertFile: etcdClientInfo.ClientCert.CertFile,
			KeyFile:  etcdClientInfo.ClientCert.KeyFile,
			CAFile:   etcdClientInfo.CA,
		},
		WrapTransport: DefaultEtcdClientTransport,
	})
	if err != nil {
		return nil, err
	}

	etcdClient := etcdclient.NewClient(etcdClientInfo.URLs)
	etcdClient.SetTransport(transport.(*http.Transport))

	for i := 0; ; i++ {
		_, err := etcdClient.Get("/", false, false)
		if err == nil || tools.IsEtcdNotFound(err) {
			break
		}
		if i > 100 {
			return nil, fmt.Errorf("could not reach etcd: %v", err)
		}
		time.Sleep(50 * time.Millisecond)
	}

	return etcdClient, nil
}

// DeleteService deletes a Service specified by its name.
func (registry *EtcdRegistry) DeleteService(name string) error {
	key := makeServiceKey(name)
	_, err := registry.etcdClient.Delete(key, true)
	if tools.IsEtcdNotFound(err) {
		return apiserver.NewNotFoundErr("service", name)
	}
	if err != nil {
		return err
	}
	key = makeServiceEndpointsKey(name)
	_, err = registry.etcdClient.Delete(key, true)
	if !tools.IsEtcdNotFound(err) {
		return err
	}
	return nil
}

func TestEtcdCreatePodWithContainersError(t *testing.T) {
	fakeClient := tools.MakeFakeEtcdClient(t)
	fakeClient.Data["/registry/hosts/machine/pods/foo"] = tools.EtcdResponseWithError{
		R: &etcd.Response{
			Node: nil,
		},
		E: tools.EtcdErrorNotFound,
	}
	fakeClient.Data["/registry/hosts/machine/kubelet"] = tools.EtcdResponseWithError{
		R: &etcd.Response{
			Node: nil,
		},
		E: tools.EtcdErrorValueRequired,
	}
	registry := MakeTestEtcdRegistry(fakeClient, []string{"machine"})
	err := registry.CreatePod("machine", api.Pod{
		JSONBase: api.JSONBase{
			ID: "foo",
		},
	})
	if err == nil {
		t.Error("Unexpected non-error")
	}
	_, err = fakeClient.Get("/registry/hosts/machine/pods/foo", false, false)
	if err == nil {
		t.Error("Unexpected non-error")
	}
	if !tools.IsEtcdNotFound(err) {
		t.Errorf("Unexpected error: %#v", err)
	}
}

// acquireOrRenewLease either races to acquire a new master lease, or update the existing master's lease
// returns true if we have the lease, and an error if one occurs.
// TODO: use the master election utility once it is merged in.
func (c *Config) acquireOrRenewLease(etcdClient *etcd.Client) (bool, error) {
	result, err := etcdClient.Get(c.key, false, false)
	if err != nil {
		if tools.IsEtcdNotFound(err) {
			// there is no current master, try to become master, create will fail if the key already exists
			_, err := etcdClient.Create(c.key, c.whoami, c.ttl)
			if err != nil {
				return false, err
			}
			c.lastLease = time.Now()
			return true, nil
		}
		return false, err
	}
	if result.Node.Value == c.whoami {
		glog.Infof("key already exists, we are the master (%s)", result.Node.Value)
		// we extend our lease @ 1/2 of the existing TTL, this ensures the master doesn't flap around
		if result.Node.Expiration.Sub(time.Now()) < time.Duration(c.ttl/2)*time.Second {
			_, err := etcdClient.CompareAndSwap(c.key, c.whoami, c.ttl, c.whoami, result.Node.ModifiedIndex)
			if err != nil {
				return false, err
			}
		}
		c.lastLease = time.Now()
		return true, nil
	}
	glog.Infof("key already exists, the master is %s, sleeping.", result.Node.Value)
	return false, nil
}

// DeleteImage deletes an existing image
func (r *Etcd) DeleteImage(id string) error {
	key := makeImageKey(id)
	err := r.Delete(key, false)
	if tools.IsEtcdNotFound(err) {
		return apierrors.NewNotFound("image", id)
	}
	return err
}

// DeleteController deletes a ReplicationController specified by its ID.
func (registry *EtcdRegistry) DeleteController(controllerID string) error {
	key := makeControllerKey(controllerID)
	_, err := registry.etcdClient.Delete(key, false)
	if tools.IsEtcdNotFound(err) {
		return apiserver.NewNotFoundErr("replicationController", controllerID)
	}
	return err
}

// DeleteController deletes a ReplicationController specified by its ID.
func (r *Registry) DeleteController(controllerID string) error {
	key := makeControllerKey(controllerID)
	err := r.Delete(key, false)
	if tools.IsEtcdNotFound(err) {
		return errors.NewNotFound("replicationController", controllerID)
	}
	return err
}

// InterpretDeleteError converts a generic etcd error on a delete
// operation into the appropriate API error.
func InterpretDeleteError(err error, kind, name string) error {
	switch {
	case tools.IsEtcdNotFound(err):
		return errors.NewNotFound(kind, name)
	default:
		return err
	}
}

// DeleteBuild deletes a Build specified by its ID.
func (r *EtcdRegistry) DeleteBuild(id string) error {
	key := makeBuildKey(id)
	err := r.Delete(key, true)
	if tools.IsEtcdNotFound(err) {
		return errors.NewNotFound("build", id)
	}
	return err
}

// DeleteImageRepository deletes an ImageRepository by id.
func (r *Etcd) DeleteImageRepository(id string) error {
	imageRepositoryKey := makeImageRepositoryKey(id)
	err := r.Delete(imageRepositoryKey, false)
	if err != nil && tools.IsEtcdNotFound(err) {
		return apierrors.NewNotFound("imageRepository", id)
	}
	return err
}

// InterpretDeleteError converts a generic etcd error on a delete
// operation into the appropriate API error.
func InterpretDeleteError(err error, kind, name string) error {
	switch {
	case tools.IsEtcdNotFound(err):
		return errors.NewNotFound(kind, name)
	case errors.IsAPIStatusError(err):
		return err
	default:
		return errors.NewInternalError(err)
	}
}

// GetBuild gets a specific Build specified by its ID.
func (r *EtcdRegistry) GetBuild(id string) (*api.Build, error) {
	var build api.Build
	err := r.ExtractObj(makeBuildKey(id), &build, false)
	if tools.IsEtcdNotFound(err) {
		return nil, errors.NewNotFound("build", id)
	}
	if err != nil {
		return nil, err
	}
	return &build, nil
}

// GetBuildConfig gets a specific BuildConfig specified by its ID.
func (r *EtcdRegistry) GetBuildConfig(id string) (*api.BuildConfig, error) {
	var config api.BuildConfig
	err := r.ExtractObj(makeBuildConfigKey(id), &config, false)
	if tools.IsEtcdNotFound(err) {
		return nil, errors.NewNotFound("buildConfig", id)
	}
	if err != nil {
		return nil, err
	}
	return &config, nil
}

// GetController gets a specific ReplicationController specified by its ID.
func (registry *EtcdRegistry) GetController(controllerID string) (*api.ReplicationController, error) {
	var controller api.ReplicationController
	key := makeControllerKey(controllerID)
	err := registry.helper().ExtractObj(key, &controller, false)
	if tools.IsEtcdNotFound(err) {
		return nil, apiserver.NewNotFoundErr("replicationController", controllerID)
	}
	if err != nil {
		return nil, err
	}
	return &controller, nil
}

// GetService obtains a Service specified by its name.
func (registry *EtcdRegistry) GetService(name string) (*api.Service, error) {
	key := makeServiceKey(name)
	var svc api.Service
	err := registry.helper().ExtractObj(key, &svc, false)
	if tools.IsEtcdNotFound(err) {
		return nil, apiserver.NewNotFoundErr("service", name)
	}
	if err != nil {
		return nil, err
	}
	return &svc, nil
}

// GetEndpoints obtains the endpoints for the service identified by 'name'.
func (r *Registry) GetEndpoints(name string) (*api.Endpoints, error) {
	key := makeServiceEndpointsKey(name)
	var endpoints api.Endpoints
	err := r.ExtractObj(key, &endpoints, false)
	if tools.IsEtcdNotFound(err) {
		return nil, apiserver.NewNotFoundErr("endpoints", name)
	}
	if err != nil {
		return nil, err
	}
	return &endpoints, nil
}