Add livestatereporter for the plugin architecture

[ffjlabo]

What this PR does:

Implement livestatereporter for the plugin architecture.

It reports the application's live state and the sync state for every registered apps by the interval of 1 minute.

Why we need it:

We need to support livestate features on the pipedv1

Which issue(s) this PR fixes:

Part of #5363

Does this PR introduce a user-facing change?:

• How are users affected by this change:
• Is this breaking change:
• How to migrate (if breaking change):

[codecov]

Codecov Report

Attention: Patch coverage is 0% with 89 lines in your changes missing coverage. Please review.

Project coverage is 26.09%. Comparing base (437d26f) to head (e9c9370).
Report is 5 commits behind head on master.

Files with missing lines	Patch %	Lines
...app/pipedv1/livestatereporter/livestatereporter.go	0.00%	88 Missing ⚠️
pkg/plugin/api/v1alpha1/client.go	0.00%	1 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##           master    #5452      +/-   ##
==========================================
- Coverage   26.23%   26.09%   -0.15%     
==========================================
  Files         452      457       +5     
  Lines       48872    49081     +209     
==========================================
- Hits        12823    12808      -15     
- Misses      35024    35251     +227     
+ Partials     1025     1022       -3     

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

[ffjlabo]

Implemented it based on this code.
https://github.com/pipe-cd/pipecd/blob/336d96d501311103dd9921bcea55722e2c8e774a/pkg/app/piped/livestatereporter/reporter.go#L104C1-L127C2

[ffjlabo]

Implemented it based on this code.
https://github.com/pipe-cd/pipecd/blob/master/pkg/app/piped/livestatereporter/kubernetes/reporter.go#L79C1-L110C2

[Warashi]

Almost LGTM. I commented on one nitpick

[Warashi]

If there are N plugins, this code sleeps for N(N+1) * 10s. It's a bit long and the expected duration seems N * 10s. One more, the info log should contain the plugin name. So, I suggest 2 ways below.

1

Suggested change

	// Avoid starting all reporters at the same time to reduce the API call burst.
	time.Sleep(time.Duration(i) * 10 * time.Second)
	r.logger.Info("starting app live state reporter for plugin")
	group.Go(func() error {
	return reporter.Run(ctx)
	})
	// Avoid starting all reporters at the same time to reduce the API call burst.
	group.Go(func() error {
	time.Sleep(time.Duration(i) * 10 * time.Second)
	reporter.logger.Info("starting app live state reporter for plugin")))
	return reporter.Run(ctx)
	})

2

Suggested change

	// Avoid starting all reporters at the same time to reduce the API call burst.
	time.Sleep(time.Duration(i) * 10 * time.Second)
	r.logger.Info("starting app live state reporter for plugin")
	group.Go(func() error {
	return reporter.Run(ctx)
	})
	// Avoid starting all reporters at the same time to reduce the API call burst.
	time.Sleep(10 * time.Second)
	reporter.logger.Info("starting app live state reporter for plugin")
	group.Go(func() error {
	return reporter.Run(ctx)
	})

[ffjlabo]

@Warashi Thanks! I fixed

• wait 10s for starting every plugin's reporter 9b697dd
• Use logger of plugin reporter 24fb566

Also, I added the info log to notice starting main reporter. a7217fc

[Warashi]

LGTM

[ffjlabo]

I noticed that we don't fix the control plane side's rpc. This is just put it as json format.
So I removed the TODO comment for ReportApplicationLiveState 🙏 . e9c9370

• pipecd/pkg/app/server/grpcapi/piped_api.go

  Lines 633 to 649 in a7217fc

  	// ReportApplicationLiveState is periodically sent to correct full state of an application.
  	// For kubernetes application, this contains a full tree of its kubernetes resources.
  	// The tree data should be written into filestore immediately and then the state in cache should be refreshsed too.
  	func (a *PipedAPI) ReportApplicationLiveState(ctx context.Context, req *pipedservice.ReportApplicationLiveStateRequest) (*pipedservice.ReportApplicationLiveStateResponse, error) {
  	_, pipedID, _, err := rpcauth.ExtractPipedToken(ctx)
  	if err != nil {
  	return nil, err
  	}
  	if err := a.validateAppBelongsToPiped(ctx, req.Snapshot.ApplicationId, pipedID); err != nil {
  	return nil, err
  	}
  	if err := a.applicationLiveStateStore.PutStateSnapshot(ctx, req.Snapshot); err != nil {
  	return nil, status.Error(codes.Internal, "failed to report application live state")
  	}
  	return &pipedservice.ReportApplicationLiveStateResponse{}, nil
  	}

• pipecd/pkg/app/server/applicationlivestatestore/store.go

  Lines 81 to 91 in a7217fc

  	func (s *store) PutStateSnapshot(ctx context.Context, snapshot *model.ApplicationLiveStateSnapshot) error {
  	if err := s.backend.Put(ctx, snapshot.ApplicationId, snapshot); err != nil {
  	s.logger.Error("failed to put application live state snapshot to filestore", zap.Error(err))
  	return err
  	}
  	if err := s.cache.Put(snapshot.ApplicationId, snapshot); err != nil {
  	s.logger.Error("failed to put application live state snapshot to cache", zap.Error(err))
  	}
  	return nil
  	}

[t-kikuc]

LGTM