Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use ListExecutors method of client Package

Best Testkube code snippet using client.ListExecutors

executor_manager.go

Source:executor_manager.go

...41	AllocateNewExec(ctx context.Context, req *pb.RegisterExecutorRequest) (*execModel.Executor, error)42	// ExecutorCount returns executor count with given status43	ExecutorCount(status model.ExecutorStatus) int44	HasExecutor(executorID string) bool45	ListExecutors() []*execModel.Executor46	GetAddr(executorID model.ExecutorID) (string, bool)47	Start(ctx context.Context) error48	Stop()49	// WatchExecutors returns a snapshot of all online executors plus50	// a stream of events describing changes that happen to the executors51	// after the snapshot is taken.52	WatchExecutors(ctx context.Context) (53		snap map[model.ExecutorID]string, stream *notifier.Receiver[model.ExecutorStatusChange], err error,54	)55	// GetExecutorInfos implements the interface scheduler.executorInfoProvider.56	// It is called by the scheduler as the source of truth for executors.57	GetExecutorInfos() map[model.ExecutorID]schedModel.ExecutorInfo58}59// ExecutorManagerImpl holds all the executors' info, including liveness, status, resource usage.60type ExecutorManagerImpl struct {61	testContext *test.Context62	wg          sync.WaitGroup63	metaClient  executormeta.Client64	mu        sync.Mutex65	executors map[model.ExecutorID]*Executor66	initHeartbeatTTL  time.Duration67	keepAliveInterval time.Duration68	rescMgr resource.RescMgr69	logRL   *rate.Limiter70	notifier *notifier.Notifier[model.ExecutorStatusChange]71}72// NewExecutorManagerImpl creates a new ExecutorManagerImpl instance73func NewExecutorManagerImpl(metaClient executormeta.Client, initHeartbeatTTL, keepAliveInterval time.Duration, ctx *test.Context) *ExecutorManagerImpl {74	return &ExecutorManagerImpl{75		testContext:       ctx,76		metaClient:        metaClient,77		executors:         make(map[model.ExecutorID]*Executor),78		initHeartbeatTTL:  initHeartbeatTTL,79		keepAliveInterval: keepAliveInterval,80		rescMgr:           resource.NewCapRescMgr(),81		logRL:             rate.NewLimiter(rate.Every(time.Second*5), 1 /*burst*/),82		notifier:          notifier.NewNotifier[model.ExecutorStatusChange](),83	}84}85// removeExecutorLocked removes an executor from the manager.86// Note that this method must be called with the lock held.87func (e *ExecutorManagerImpl) removeExecutorLocked(id model.ExecutorID) error {88	log.Info("begin to remove executor", zap.String("id", string(id)))89	exec, ok := e.executors[id]90	if !ok {91		// This executor has been removed92		return errors.ErrUnknownExecutorID.GenWithStackByArgs(id)93	}94	addr := exec.Address95	delete(e.executors, id)96	e.rescMgr.Unregister(id)97	log.Info("notify to offline exec")98	if test.GetGlobalTestFlag() {99		e.testContext.NotifyExecutorChange(&test.ExecutorChangeEvent{100			Tp:   test.Delete,101			Time: time.Now(),102		})103	}104	e.notifier.Notify(model.ExecutorStatusChange{105		ID:   id,106		Tp:   model.EventExecutorOffline,107		Addr: addr,108	})109	// Delete the executor from the database. This operation may take a long time,110	// and it may fail if the database is unavailable. So we don't want to hold the lock.111	// Instead, we do the deletion in a goroutine.112	//113	// We use ttl mechanism to manage the executor's life cycle. So we can tolerate114	// that a tombstone executor may be left in the database.115	e.wg.Add(1)116	go func() {117		defer e.wg.Done()118		ctx, cancel := context.WithTimeout(context.Background(), time.Second*5)119		defer cancel()120		if err := e.metaClient.DeleteExecutor(ctx, id); err != nil {121			log.Warn("failed to delete executor from database", zap.String("id", string(id)), zap.Error(err))122		}123	}()124	return nil125}126// HandleHeartbeat implements pb interface,127func (e *ExecutorManagerImpl) HandleHeartbeat(req *pb.HeartbeatRequest) (*pb.HeartbeatResponse, error) {128	if e.logRL.Allow() {129		log.Info("handle heart beat", zap.Stringer("req", req))130	}131	e.mu.Lock()132	execID := model.ExecutorID(req.ExecutorId)133	exec, ok := e.executors[execID]134	// executor not exists135	if !ok {136		e.mu.Unlock()137		err := errors.ErrUnknownExecutorID.FastGenByArgs(req.ExecutorId)138		return &pb.HeartbeatResponse{Err: errors.ToPBError(err)}, nil139	}140	e.mu.Unlock()141	status := model.ExecutorStatus(req.Status)142	if err := exec.heartbeat(req.Ttl, status); err != nil {143		return &pb.HeartbeatResponse{Err: errors.ToPBError(err)}, nil144	}145	usage := model.RescUnit(req.GetResourceUsage())146	if err := e.rescMgr.Update(execID, usage, usage, status); err != nil {147		return nil, err148	}149	resp := &pb.HeartbeatResponse{}150	return resp, nil151}152// registerExec registers executor to both executor manager and resource manager.153// Note that this method must be called with the lock held.154func (e *ExecutorManagerImpl) registerExecLocked(executorMeta *execModel.Executor) {155	log.Info("register executor", zap.Any("executor", executorMeta))156	exec := &Executor{157		Executor:       *executorMeta,158		lastUpdateTime: time.Now(),159		heartbeatTTL:   e.initHeartbeatTTL,160		status:         model.Initing,161		logRL:          rate.NewLimiter(rate.Every(time.Second*5), 1 /*burst*/),162	}163	e.executors[executorMeta.ID] = exec164	e.notifier.Notify(model.ExecutorStatusChange{165		ID:   executorMeta.ID,166		Tp:   model.EventExecutorOnline,167		Addr: executorMeta.Address,168	})169	e.rescMgr.Register(exec.ID, exec.Address, model.RescUnit(exec.Capability))170}171// AllocateNewExec allocates new executor info to a give RegisterExecutorRequest172// and then registers the executor.173func (e *ExecutorManagerImpl) AllocateNewExec(ctx context.Context, req *pb.RegisterExecutorRequest) (*execModel.Executor, error) {174	pbExecutor := req.Executor175	log.Info("allocate new executor", zap.Stringer("executor", pbExecutor))176	e.mu.Lock()177	var executorID model.ExecutorID178	for {179		executorID = generateExecutorID(pbExecutor.GetName())180		if _, ok := e.executors[executorID]; !ok {181			break182		}183	}184	labelSet, err := label.NewSetFromMap(req.GetExecutor().GetLabels())185	if err != nil {186		return nil, err187	}188	executorMeta := &execModel.Executor{189		ID:         executorID,190		Name:       pbExecutor.GetName(),191		Address:    pbExecutor.GetAddress(),192		Capability: int(pbExecutor.GetCapability()),193		Labels:     execModel.LabelSet(labelSet),194	}195	e.registerExecLocked(executorMeta)196	e.mu.Unlock()197	// Store the executor info to database.198	// If any error occurs, client shouldn't use the executor.199	// The executor in the map will be removed after the ttl expires.200	if err := e.metaClient.CreateExecutor(ctx, executorMeta); err != nil {201		return nil, perrors.Trace(err)202	}203	return executorMeta, nil204}205func generateExecutorID(name string) model.ExecutorID {206	val := rand.Uint32()207	id := fmt.Sprintf("%s-%08x", name, val)208	return model.ExecutorID(id)209}210// HasExecutor implements ExecutorManager.HasExecutor211func (e *ExecutorManagerImpl) HasExecutor(executorID string) bool {212	e.mu.Lock()213	defer e.mu.Unlock()214	_, ok := e.executors[model.ExecutorID(executorID)]215	return ok216}217// ListExecutors implements ExecutorManager.ListExecutors218func (e *ExecutorManagerImpl) ListExecutors() []*execModel.Executor {219	e.mu.Lock()220	defer e.mu.Unlock()221	ret := make([]*execModel.Executor, 0, len(e.executors))222	for _, exec := range e.executors {223		execMeta := exec.Executor224		ret = append(ret, &execMeta)225	}226	return ret227}228// Executor records the status of an executor instance.229type Executor struct {230	execModel.Executor231	status model.ExecutorStatus232	mu sync.RWMutex...

discovery_client.go

Source:discovery_client.go

...26	RegisterExecutor(27		ctx context.Context,28		request *enginepb.RegisterExecutorRequest,29	) (model.ExecutorID, error)30	// ListExecutors lists all executors.31	ListExecutors(ctx context.Context) ([]*enginepb.Executor, error)32	// ListMasters lists all masters.33	ListMasters(ctx context.Context) ([]*enginepb.Master, error)34	// Heartbeat sends a heartbeat message to the server.35	Heartbeat(36		ctx context.Context,37		request *enginepb.HeartbeatRequest,38	) (*enginepb.HeartbeatResponse, error)39	// RegisterMetaStore registers a new metastore.40	// Deprecated41	RegisterMetaStore(42		ctx context.Context,43		request *enginepb.RegisterMetaStoreRequest,44	) error45	// QueryMetaStore queries the details of a metastore.46	QueryMetaStore(47		ctx context.Context,48		request *enginepb.QueryMetaStoreRequest,49	) (*enginepb.QueryMetaStoreResponse, error)50}51var _ DiscoveryClient = &discoveryClient{}52type discoveryClient struct {53	cli enginepb.DiscoveryClient54}55// NewDiscoveryClient returns a DiscoveryClient.56func NewDiscoveryClient(cli enginepb.DiscoveryClient) DiscoveryClient {57	return &discoveryClient{cli: cli}58}59func (c *discoveryClient) RegisterExecutor(60	ctx context.Context,61	request *enginepb.RegisterExecutorRequest,62) (model.ExecutorID, error) {63	var ret model.ExecutorID64	err := retry.Do(ctx, func() error {65		call := internal.NewCall(66			c.cli.RegisterExecutor,67			request,68			// RegisterExecutor is not idempotent in general69			// TODO review idempotency70			// internal.WithForceNoRetry()71		)72		executor, err := call.Do(ctx)73		if err != nil {74			return err75		}76		ret = model.ExecutorID(executor.Id)77		return nil78	})79	if err != nil {80		return "", errors.Trace(err)81	}82	return ret, nil83}84func (c *discoveryClient) ListExecutors(ctx context.Context) ([]*enginepb.Executor, error) {85	call := internal.NewCall(c.cli.ListExecutors, &enginepb.ListExecutorsRequest{})86	resp, err := call.Do(ctx)87	if err != nil {88		return nil, err89	}90	return resp.Executors, nil91}92func (c *discoveryClient) ListMasters(ctx context.Context) ([]*enginepb.Master, error) {93	call := internal.NewCall(c.cli.ListMasters, &enginepb.ListMastersRequest{})94	resp, err := call.Do(ctx)95	if err != nil {96		return nil, err97	}98	return resp.Masters, nil99}...

executor.go

Source:executor.go

...18func (c ExecutorClient) GetExecutor(name string) (executor testkube.ExecutorDetails, err error) {19	uri := c.executorTransport.GetURI("/executors/%s", name)20	return c.executorTransport.Execute(http.MethodGet, uri, nil, nil)21}22// ListExecutors list all executors23func (c ExecutorClient) ListExecutors(selector string) (executors testkube.ExecutorsDetails, err error) {24	uri := c.executorTransport.GetURI("/executors")25	params := map[string]string{26		"selector": selector,27	}28	return c.executorTransport.ExecuteMultiple(http.MethodGet, uri, nil, params)29}30// CreateExecutor creates new Executor Custom Resource31func (c ExecutorClient) CreateExecutor(options CreateExecutorOptions) (executor testkube.ExecutorDetails, err error) {32	uri := c.executorTransport.GetURI("/executors")33	request := testkube.ExecutorCreateRequest(options)34	body, err := json.Marshal(request)35	if err != nil {36		return executor, err37	}...

ListExecutors

Using AI Code Generation

1import (2func main() {3	masterClient := client.New(httpcli.New(4		httpcli.Codec(encoding.ProtobufCodec),5	agentClient := client.New(httpcli.New(6		httpcli.Codec(encoding.ProtobufCodec),7	schedulerClient := client.New(httpcli.New(8		httpcli.Codec(encoding.ProtobufCodec),9	executorClient := client.New(httpcli.New(10		httpcli.Codec(encoding.ProtobufCodec),11	operatorClient := client.New(httpcli.New(12		httpcli.Codec(encoding.ProtobufCodec),13	executors, err := masterClient.ListExecutors(context.Background())14	if err != nil {15		panic(err)16	}17	fmt.Println(executors)18}19import (

ListExecutors

Using AI Code Generation

1import (2func main() {3	c := client.New(4		client.Config{5		},6	call := calls.ListExecutors(requests.ListExecutors{7		Filters: structs.CallListExecutorsFilters{8			FrameworkID: structs.FrameworkID{9			},10			ExecutorID: structs.ExecutorID{11			},12			SlaveID: structs.SlaveID{13			},14		},15	})16	resp, err := c.Call(call)17	if err != nil {18		fmt.Printf("Error: %v", err)19	}20	if err := resp.Validate(); err != nil {21		fmt.Printf("Error: %v", err)22	}23	if err := resp.Error(); err != nil {24		fmt.Printf("Error: %v", err)25	}26	if listExecutors, ok := resp.Response.(*responses.ListExecutors); ok {

ListExecutors

Using AI Code Generation

1import (2func main() {3	if err != nil {4		log.Fatal(err)5	}6	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)7	defer cancel()8	executors, resp, err := c.ListExecutors(ctx, "agent1", "framework1", "executor1")9	if err != nil {10		log.Fatal(err)11	}12	fmt.Println(executors)13	fmt.Println(resp)14}

ListExecutors

Using AI Code Generation

1import (2func main() {3	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)4	defer cancel()5	client, err := master.NewClient(ctx, master.ClientConfig{6		HTTP: httpcli.New(7		Codec: encoding.ProtoCodec(json.NewCodec()),8	})9	if err != nil {10		log.Fatal(err)11	}12	executors, err := client.ListExecutors(ctx, master.Call_ListExecutors{})13	if err != nil {14		log.Fatal(err)15	}16	fmt.Printf("Found %d executors:17", len(executors))18	for _, executor := range executors {19		fmt.Printf("ExecutorID: %s20		fmt.Printf("FrameworkID: %s21		fmt.Printf("FrameworkName: %s22		fmt.Printf("SlaveID: %s23		fmt.Printf("Source: %s24		fmt.Printf("Container: %s25		fmt.Printf("Labels: %s26		fmt.Printf("Resources: %s27		fmt.Printf("ExecutorInfo: %s28		fmt.Printf("Status: %s29		fmt.Printf("CompletionTime: %s30		fmt.Printf("QueuedTasks: %s31		fmt.Printf("CompletedTasks: %s32		fmt.Printf("Terminated

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.