// objectMetaAndKind retrieves kind and ObjectMeta from a runtime object, or returns an error.
func objectMetaAndKind(typer runtime.ObjectTyper, obj runtime.Object) (*api.ObjectMeta, string, error) {
	objectMeta, err := api.ObjectMetaFor(obj)
	if err != nil {
		return nil, "", errors.NewInternalError(err)
	}
	_, kind, err := typer.ObjectVersionAndKind(obj)
	if err != nil {
		return nil, "", errors.NewInternalError(err)
	}
	return objectMeta, kind, nil
}

// TestHandle_nonfatalCreateError ensures that a failed API attempt to create
// a new deployment for an updated config results in a nonfatal error.
func TestHandle_nonfatalCreateError(t *testing.T) {
	configController := &DeploymentConfigController{
		makeDeployment: func(config *deployapi.DeploymentConfig) (*kapi.ReplicationController, error) {
			return deployutil.MakeDeployment(config, api.Codec)
		},
		deploymentClient: &deploymentClientImpl{
			createDeploymentFunc: func(namespace string, deployment *kapi.ReplicationController) (*kapi.ReplicationController, error) {
				return nil, kerrors.NewInternalError(fmt.Errorf("test error"))
			},
			listDeploymentsForConfigFunc: func(namespace, configName string) (*kapi.ReplicationControllerList, error) {
				return &kapi.ReplicationControllerList{}, nil
			},
			updateDeploymentFunc: func(namespace string, deployment *kapi.ReplicationController) (*kapi.ReplicationController, error) {
				t.Fatalf("unexpected update call with deployment %v", deployment)
				return nil, nil
			},
		},
		recorder: &record.FakeRecorder{},
	}

	err := configController.Handle(deploytest.OkDeploymentConfig(1))
	if err == nil {
		t.Fatalf("expected error")
	}
	if _, isFatal := err.(fatalError); isFatal {
		t.Fatalf("expected a nonfatal error, got a fatal error: %v", err)
	}
}

func TestCreateGeneratorError(t *testing.T) {
	rest := REST{
		generator: Client{
			GRFn: func(from, to *deployapi.DeploymentConfig, spec *deployapi.DeploymentConfigRollbackSpec) (*deployapi.DeploymentConfig, error) {
				return nil, kerrors.NewInternalError(fmt.Errorf("something terrible happened"))
			},
			RCFn: func(ctx kapi.Context, name string) (*kapi.ReplicationController, error) {
				deployment, _ := deployutil.MakeDeployment(deploytest.OkDeploymentConfig(1), kapi.Codec)
				return deployment, nil
			},
			DCFn: func(ctx kapi.Context, name string) (*deployapi.DeploymentConfig, error) {
				return deploytest.OkDeploymentConfig(1), nil
			},
		},
		codec: api.Codec,
	}

	_, err := rest.Create(kapi.NewDefaultContext(), &deployapi.DeploymentConfigRollback{
		Spec: deployapi.DeploymentConfigRollbackSpec{
			From: kapi.ObjectReference{
				Name:      "deployment",
				Namespace: kapi.NamespaceDefault,
			},
		},
	})

	if err == nil || !strings.Contains(err.Error(), "something terrible happened") {
		t.Errorf("Unexpected error: %v", err)
	}
}

// TestHandle_fatalError ensures that in internal (not API) failure to make a
// deployment from an updated config results in a fatal error.
func TestHandle_fatalError(t *testing.T) {
	configController := &DeploymentConfigController{
		makeDeployment: func(config *deployapi.DeploymentConfig) (*kapi.ReplicationController, error) {
			return nil, fmt.Errorf("couldn't make deployment")
		},
		deploymentClient: &deploymentClientImpl{
			createDeploymentFunc: func(namespace string, deployment *kapi.ReplicationController) (*kapi.ReplicationController, error) {
				t.Fatalf("unexpected call to create")
				return nil, kerrors.NewInternalError(fmt.Errorf("test error"))
			},
			listDeploymentsForConfigFunc: func(namespace, configName string) (*kapi.ReplicationControllerList, error) {
				return &kapi.ReplicationControllerList{}, nil
			},
			updateDeploymentFunc: func(namespace string, deployment *kapi.ReplicationController) (*kapi.ReplicationController, error) {
				t.Fatalf("unexpected update call with deployment %v", deployment)
				return nil, nil
			},
		},
	}

	err := configController.Handle(deploytest.OkDeploymentConfig(1))
	if err == nil {
		t.Fatalf("expected error")
	}
	if _, isFatal := err.(fatalError); !isFatal {
		t.Fatalf("expected a fatal error, got: %v", err)
	}
}

// InterpretUpdateError converts a generic etcd error on a update
// operation into the appropriate API error.
func InterpretUpdateError(err error, kind, name string) error {
	switch {
	case tools.IsEtcdTestFailed(err), tools.IsEtcdNodeExist(err):
		return errors.NewConflict(kind, name, err)
	case errors.IsAPIStatusError(err):
		return err
	default:
		return errors.NewInternalError(err)
	}
}

// InterpretCreateError converts a generic etcd error on a create
// operation into the appropriate API error.
func InterpretCreateError(err error, kind, name string) error {
	switch {
	case tools.IsEtcdNodeExist(err):
		return errors.NewAlreadyExists(kind, name)
	case errors.IsAPIStatusError(err):
		return err
	default:
		return errors.NewInternalError(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)
	}
}

func (rs *REST) Delete(ctx api.Context, name string) (runtime.Object, error) {
	obj, err := rs.registry.Get(ctx, name)
	if err != nil {
		return nil, err
	}
	_, ok := obj.(*api.Event)
	if !ok {
		return nil, errors.NewInternalError(fmt.Errorf("stored object %s is not of type event: %#v", name, obj))
	}
	return rs.registry.Delete(ctx, name, nil)
}

// ServeHTTP implements rest.HookHandler
func (c *controller) ServeHTTP(w http.ResponseWriter, req *http.Request, ctx kapi.Context, name, subpath string) error {
	parts := strings.Split(subpath, "/")
	if len(parts) < 2 {
		return errors.NewBadRequest(fmt.Sprintf("unexpected hook subpath %s", subpath))
	}
	secret, hookType := parts[0], parts[1]

	plugin, ok := c.plugins[hookType]
	if !ok {
		return errors.NewNotFound("BuildConfigHook", hookType)
	}

	config, err := c.registry.GetBuildConfig(ctx, name)
	if err != nil {
		// clients should not be able to find information about build configs in the system unless the config exists
		// and the secret matches
		return errors.NewUnauthorized(fmt.Sprintf("the webhook %q for %q did not accept your secret", hookType, name))
	}

	revision, proceed, err := plugin.Extract(config, secret, "", req)
	switch err {
	case webhook.ErrSecretMismatch, webhook.ErrHookNotEnabled:
		return errors.NewUnauthorized(fmt.Sprintf("the webhook %q for %q did not accept your secret", hookType, name))
	case nil:
	default:
		return errors.NewInternalError(fmt.Errorf("hook failed: %v", err))
	}

	if !proceed {
		return nil
	}

	request := &buildapi.BuildRequest{
		ObjectMeta: kapi.ObjectMeta{Name: name},
		Revision:   revision,
	}
	if _, err := c.instantiator.Instantiate(config.Namespace, request); err != nil {
		return errors.NewInternalError(fmt.Errorf("could not generate a build: %v", err))
	}
	return nil
}

// getMatchingSecurityContextConstraints returns constraints from the store that match the group,
// uid, or user of the service account.
func getMatchingSecurityContextConstraints(store cache.Store, userInfo user.Info) ([]*kapi.SecurityContextConstraints, error) {
	matchedConstraints := make([]*kapi.SecurityContextConstraints, 0)

	for _, c := range store.List() {
		constraint, ok := c.(*kapi.SecurityContextConstraints)
		if !ok {
			return nil, errors.NewInternalError(fmt.Errorf("error converting object from store to a security context constraint: %v", c))
		}
		if constraintAppliesTo(constraint, userInfo) {
			matchedConstraints = append(matchedConstraints, constraint)
		}
	}

	return matchedConstraints, nil
}

// ValidateService tests if required fields in the service are set.
func ValidateService(service *api.Service, lister ServiceLister, ctx api.Context) errs.ValidationErrorList {
	allErrs := errs.ValidationErrorList{}
	if len(service.Name) == 0 {
		allErrs = append(allErrs, errs.NewFieldRequired("name", service.Name))
	} else if !util.IsDNS952Label(service.Name) {
		allErrs = append(allErrs, errs.NewFieldInvalid("name", service.Name, ""))
	}
	if !util.IsDNSSubdomain(service.Namespace) {
		allErrs = append(allErrs, errs.NewFieldInvalid("namespace", service.Namespace, ""))
	}
	if !util.IsValidPortNum(service.Spec.Port) {
		allErrs = append(allErrs, errs.NewFieldInvalid("spec.port", service.Spec.Port, ""))
	}
	if len(service.Spec.Protocol) == 0 {
		service.Spec.Protocol = "TCP"
	} else if !supportedPortProtocols.Has(strings.ToUpper(string(service.Spec.Protocol))) {
		allErrs = append(allErrs, errs.NewFieldNotSupported("spec.protocol", service.Spec.Protocol))
	}

	if service.Spec.Selector != nil {
		allErrs = append(allErrs, validateLabels(service.Spec.Selector, "spec.selector")...)
	}
	allErrs = append(allErrs, validateLabels(service.Labels, "labels")...)

	if service.Spec.CreateExternalLoadBalancer {
		services, err := lister.ListServices(ctx)
		if err != nil {
			allErrs = append(allErrs, errs.NewInternalError(err))
		} else {
			for i := range services.Items {
				if services.Items[i].Name != service.Name &&
					services.Items[i].Spec.CreateExternalLoadBalancer &&
					services.Items[i].Spec.Port == service.Spec.Port {
					allErrs = append(allErrs, errs.NewConflict("service", service.Name, fmt.Errorf("port: %d is already in use", service.Spec.Port)))
					break
				}
			}
		}
	}
	if service.Spec.SessionAffinity == "" {
		service.Spec.SessionAffinity = api.AffinityTypeNone
	} else if !supportedSessionAffinityType.Has(string(service.Spec.SessionAffinity)) {
		allErrs = append(allErrs, errs.NewFieldNotSupported("spec.sessionAffinity", service.Spec.SessionAffinity))
	}

	return allErrs
}

func (rs *REST) Create(ctx api.Context, obj runtime.Object) (runtime.Object, error) {
	event, ok := obj.(*api.Event)
	if !ok {
		return nil, errors.NewInternalError(fmt.Errorf("received object is not of type event: %#v", obj))
	}
	if api.NamespaceValue(ctx) != "" {
		if !api.ValidNamespace(ctx, &event.ObjectMeta) {
			return nil, errors.NewConflict("event", event.Namespace, fmt.Errorf("event.namespace does not match the provided context"))
		}
	}
	if errs := validation.ValidateEvent(event); len(errs) > 0 {
		return nil, errors.NewInvalid("event", event.Name, errs)
	}
	api.FillObjectMetaSystemFields(ctx, &event.ObjectMeta)

	err := rs.registry.CreateWithName(ctx, event.Name, event)
	if err != nil {
		return nil, err
	}
	return rs.registry.Get(ctx, event.Name)
}

// TestHandle_cleanupPodFail ensures that a failed attempt to clean up the
// deployer pod for a completed deployment results in a nonfatal error.
func TestHandle_cleanupPodFail(t *testing.T) {
	controller := &DeploymentController{
		decodeConfig: func(deployment *kapi.ReplicationController) (*deployapi.DeploymentConfig, error) {
			return deployutil.DecodeDeploymentConfig(deployment, api.Codec)
		},
		deploymentClient: &deploymentClientImpl{
			updateDeploymentFunc: func(namespace string, deployment *kapi.ReplicationController) (*kapi.ReplicationController, error) {
				t.Fatalf("unexpected deployment update")
				return nil, nil
			},
		},
		podClient: &podClientImpl{
			createPodFunc: func(namespace string, pod *kapi.Pod) (*kapi.Pod, error) {
				t.Fatalf("unexpected call to create pod")
				return nil, nil
			},
			deletePodFunc: func(namespace, name string) error {
				return kerrors.NewInternalError(fmt.Errorf("test error"))
			},
			getDeployerPodsForFunc: func(namespace, name string) ([]kapi.Pod, error) {
				return []kapi.Pod{{}}, nil
			},
		},
		makeContainer: func(strategy *deployapi.DeploymentStrategy) (*kapi.Container, error) {
			t.Fatalf("unexpected call to make container")
			return nil, nil
		},
		recorder: &record.FakeRecorder{},
	}

	// Verify error
	config := deploytest.OkDeploymentConfig(1)
	deployment, _ := deployutil.MakeDeployment(config, kapi.Codec)
	deployment.Annotations[deployapi.DeploymentStatusAnnotation] = string(deployapi.DeploymentStatusComplete)
	err := controller.Handle(deployment)

	if err == nil {
		t.Fatalf("expected an error")
	}
}

func (rs *REST) getAttrs(obj runtime.Object) (objLabels labels.Set, objFields fields.Set, err error) {
	event, ok := obj.(*api.Event)
	if !ok {
		return nil, nil, errors.NewInternalError(fmt.Errorf("object is not of type event: %#v", obj))
	}
	l := event.Labels
	if l == nil {
		l = labels.Set{}
	}
	return l, fields.Set{
		"metadata.name":                  event.Name,
		"involvedObject.kind":            event.InvolvedObject.Kind,
		"involvedObject.namespace":       event.InvolvedObject.Namespace,
		"involvedObject.name":            event.InvolvedObject.Name,
		"involvedObject.uid":             string(event.InvolvedObject.UID),
		"involvedObject.apiVersion":      event.InvolvedObject.APIVersion,
		"involvedObject.resourceVersion": fmt.Sprintf("%s", event.InvolvedObject.ResourceVersion),
		"involvedObject.fieldPath":       event.InvolvedObject.FieldPath,
		"reason":                         event.Reason,
		"source":                         event.Source.Component,
	}, nil
}

// Create registers a given new Route instance to rs.registry.
func (rs *REST) Create(ctx kapi.Context, obj runtime.Object) (runtime.Object, error) {
	route, ok := obj.(*api.Route)
	if !ok {
		return nil, errors.NewBadRequest(fmt.Sprintf("not a route: %#v", obj))
	}
	if !kapi.ValidNamespace(ctx, &route.ObjectMeta) {
		return nil, errors.NewConflict("route", route.Namespace, fmt.Errorf("Route.Namespace does not match the provided context"))
	}

	shard, err := rs.allocator.AllocateRouterShard(route)
	if err != nil {
		return nil, errors.NewInternalError(fmt.Errorf("allocation error: %s for route: %#v", err, obj))
	}

	if route.Annotations == nil {
		route.Annotations = map[string]string{}
	}
	if len(route.Host) == 0 {
		route.Host = rs.allocator.GenerateHostname(route, shard)
		route.Annotations[HostGeneratedAnnotationKey] = "true"
	} else {
		route.Annotations[HostGeneratedAnnotationKey] = "false"
	}

	if errs := validation.ValidateRoute(route); len(errs) > 0 {
		return nil, errors.NewInvalid("route", route.Name, errs)
	}
	if len(route.Name) == 0 {
		route.Name = uuid.NewUUID().String()
	}

	kapi.FillObjectMetaSystemFields(ctx, &route.ObjectMeta)

	err = rs.registry.CreateRoute(ctx, route)
	if err != nil {
		return nil, err
	}
	return rs.registry.GetRoute(ctx, route.Name)
}

import (
	"fmt"
	"sync"

	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
	k8serr "github.com/GoogleCloudPlatform/kubernetes/pkg/api/errors"
	etcderr "github.com/GoogleCloudPlatform/kubernetes/pkg/api/errors/etcd"
	"github.com/GoogleCloudPlatform/kubernetes/pkg/registry/service"
	"github.com/GoogleCloudPlatform/kubernetes/pkg/registry/service/allocator"
	"github.com/GoogleCloudPlatform/kubernetes/pkg/runtime"
	"github.com/GoogleCloudPlatform/kubernetes/pkg/tools"
)

var (
	// Placeholder error that should not be surfaced to the user.
	errorUnableToAllocate = k8serr.NewInternalError(fmt.Errorf("unable to allocate"))
)

// Etcd exposes a service.Allocator that is backed by etcd.
// TODO: allow multiple allocations to be tried at once
// TODO: subdivide the keyspace to reduce conflicts
// TODO: investigate issuing a CAS without reading first
type Etcd struct {
	lock sync.Mutex

	alloc  allocator.Snapshottable
	helper tools.EtcdHelper
	last   string

	baseKey string
	kind    string

func (r *REST) Create(ctx kapi.Context, obj runtime.Object) (runtime.Object, error) {

	if err := rest.BeforeCreate(projectrequestregistry.Strategy, ctx, obj); err != nil {
		return nil, err
	}

	projectRequest := obj.(*projectapi.ProjectRequest)

	if _, err := r.openshiftClient.Projects().Get(projectRequest.Name); err == nil {
		return nil, kapierror.NewAlreadyExists("project", projectRequest.Name)
	}

	projectName := projectRequest.Name
	projectAdmin := ""
	if userInfo, exists := kapi.UserFrom(ctx); exists {
		projectAdmin = userInfo.GetName()
	}

	template, err := r.getTemplate()
	if err != nil {
		return nil, err
	}

	for i := range template.Parameters {
		switch template.Parameters[i].Name {
		case ProjectAdminUserParam:
			template.Parameters[i].Value = projectAdmin
		case ProjectDescriptionParam:
			template.Parameters[i].Value = projectRequest.Description
		case ProjectDisplayNameParam:
			template.Parameters[i].Value = projectRequest.DisplayName
		case ProjectNameParam:
			template.Parameters[i].Value = projectName
		}
	}

	list, err := r.openshiftClient.TemplateConfigs(kapi.NamespaceDefault).Create(template)
	if err != nil {
		return nil, err
	}
	if err := utilerrors.NewAggregate(runtime.DecodeList(list.Objects, kapi.Scheme)); err != nil {
		return nil, kapierror.NewInternalError(err)
	}

	// one of the items in this list should be the project.  We are going to locate it, remove it from the list, create it separately
	var projectFromTemplate *projectapi.Project
	objectsToCreate := &kapi.List{}
	for i := range list.Objects {
		if templateProject, ok := list.Objects[i].(*projectapi.Project); ok {
			projectFromTemplate = templateProject

			if len(list.Objects) > (i + 1) {
				objectsToCreate.Items = append(objectsToCreate.Items, list.Objects[i+1:]...)
			}
			break
		}

		objectsToCreate.Items = append(objectsToCreate.Items, list.Objects[i])
	}
	if projectFromTemplate == nil {
		return nil, kapierror.NewInternalError(fmt.Errorf("the project template (%s/%s) is not correctly configured: must contain a project resource", r.templateNamespace, r.templateName))
	}

	// we split out project creation separately so that in a case of racers for the same project, only one will win and create the rest of their template objects
	if _, err := r.openshiftClient.Projects().Create(projectFromTemplate); err != nil {
		return nil, err
	}

	bulk := configcmd.Bulk{
		Mapper: latest.RESTMapper,
		Typer:  kapi.Scheme,
		RESTClientFactory: func(mapping *meta.RESTMapping) (resource.RESTClient, error) {
			return r.openshiftClient, nil
		},
	}
	if err := utilerrors.NewAggregate(bulk.Create(objectsToCreate, projectName)); err != nil {
		return nil, kapierror.NewInternalError(err)
	}

	return r.openshiftClient.Projects().Get(projectName)
}