Method StartFunction

fs/manager.go:195–248

(f *model.Function)

}

func (fm *functionManagerImpl) StartFunction(f *model.Function) error { // TODO: Shouldn't use pointer here
	if err := f.Validate(); err != nil {
		return err
	}
	fm.functionsLock.Lock()
	if _, exist := fm.functions[GetNamespacedName(f.Namespace, f.Name)]; exist {
		fm.functionsLock.Unlock()
		return common.ErrorFunctionExists
	}
	fm.functions[GetNamespacedName(f.Namespace, f.Name)] = make([]api.FunctionInstance, f.Replicas)
	fm.functionsLock.Unlock()
	funcCtx := fm.createFuncCtx()
	for i := int32(0); i < f.Replicas; i++ {
		runtimeType := f.Runtime.Type

		instanceLogger := fm.log.SubLogger("functionName", f.Name, "instanceIndex", int(i), "runtimeType", runtimeType)
		instance := fm.options.instanceFactory.NewFunctionInstance(f, funcCtx, i, instanceLogger)
		fm.functionsLock.Lock()
		fm.functions[GetNamespacedName(f.Namespace, f.Name)][i] = instance
		fm.functionsLock.Unlock()
		runtimeFactory, err := fm.getRuntimeFactory(runtimeType)
		if err != nil {
			return err
		}
		var sources []<-chan contube.Record
		for _, t := range f.Sources {
			sourceFactory, err := fm.getTubeFactory(&t)
			if err != nil {
				return nil
			}
			sourceChan, err := sourceFactory.NewSourceTube(instance.Context(), t.Config)
			if err != nil {
				return fmt.Errorf("failed to create source event queue: %w", err)
			}
			sources = append(sources, sourceChan)
		}
		sinkFactory, err := fm.getTubeFactory(&f.Sink)
		if err != nil {
			return nil
		}
		sink, err := sinkFactory.NewSinkTube(instance.Context(), f.Sink.Config)
		if err != nil {
			return fmt.Errorf("failed to create sink event queue: %w", err)
		}

		runtime, err := runtimeFactory.NewFunctionRuntime(instance)
		if err != nil {
			return fmt.Errorf("failed to create runtime: %w", err)
		}

		go instance.Run(runtime, sources, sink)
	}
	return nil
}

func (fm *functionManagerImpl) DeleteFunction(namespace, name string) error {
	fm.functionsLock.Lock()