Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

support full sync when the work watcher creating or the work client reconnected #133

Merged
merged 1 commit into from
Jun 21, 2024
Merged

support full sync when the work watcher creating or the work client reconnected #133

merged 1 commit into from
Jun 21, 2024

Conversation

skeeey
Copy link
Contributor

@skeeey skeeey commented Jun 18, 2024

refer to #131

@skeeey
Copy link
Contributor Author

skeeey commented Jun 18, 2024

/cc @qiujian16 @clyang82 @machi1990

@skeeey skeeey mentioned this pull request Jun 19, 2024
Merged
@skeeey
Copy link
Contributor Author

skeeey commented Jun 20, 2024

make test
OCM_ENV=testing gotestsum --jsonfile-timing-events=/home/cloud-user/go/src/github.com/openshift-online/maestro/unit-test-results.json --format short-verbose -- -p 1 -v
./pkg/...
./cmd/...
PASS pkg/api.TestDecodeManifestBundle/empty (0.00s)
PASS pkg/api.TestDecodeManifestBundle/valid (0.00s)
PASS pkg/api.TestDecodeManifestBundle (0.00s)
PASS pkg/api.TestDecodeManifestBundleToObjects/empty (0.00s)
PASS pkg/api.TestDecodeManifestBundleToObjects/valid (0.00s)
PASS pkg/api.TestDecodeManifestBundleToObjects (0.00s)
PASS pkg/api.TestDecodeBundleStatus/empty (0.00s)
PASS pkg/api.TestDecodeBundleStatus/valid (0.00s)
PASS pkg/api.TestDecodeBundleStatus (0.00s)
PASS pkg/api.TestEncodeManifest/empty (0.00s)
PASS pkg/api.TestEncodeManifest/valid (0.00s)
PASS pkg/api.TestEncodeManifest/valid#01 (0.00s)
PASS pkg/api.TestEncodeManifest (0.00s)
PASS pkg/api.TestDecodeManifest/empty (0.00s)
PASS pkg/api.TestDecodeManifest/valid (0.00s)
PASS pkg/api.TestDecodeManifest (0.00s)
PASS pkg/api.TestDecodeStatus/empty (0.00s)
PASS pkg/api.TestDecodeStatus/valid (0.00s)
PASS pkg/api.TestDecodeStatus (0.00s)
PASS pkg/api
EMPTY pkg/api/openapi
EMPTY pkg/api/presenters
EMPTY pkg/auth
EMPTY pkg/client/cloudevents
PASS pkg/client/cloudevents/grpcsource.TestToManifestWork/covert_a_resource_bundle_-has_empty_fields (0.00s)
PASS pkg/client/cloudevents/grpcsource.TestToManifestWork/covert_a_resource_bundle (0.00s)
PASS pkg/client/cloudevents/grpcsource.TestToManifestWork (0.00s)
PASS pkg/client/cloudevents/grpcsource
EMPTY pkg/client/ocm
PASS pkg/config.TestConfigReadStringFile (0.00s)
PASS pkg/config.TestConfigReadIntFile (0.00s)
PASS pkg/config.TestConfigReadBoolFile (0.00s)
PASS pkg/config.TestConfigReadQuotedFile (0.00s)
PASS pkg/config
EMPTY pkg/constants
PASS pkg/controllers.TestControllerFramework (0.00s)
PASS pkg/controllers
EMPTY pkg/dao
EMPTY pkg/dao/mocks
EMPTY pkg/db
EMPTY pkg/db/db_context
EMPTY pkg/db/db_session
EMPTY pkg/db/migrations
EMPTY pkg/db/mocks
EMPTY pkg/db/transaction
EMPTY pkg/dispatcher
PASS pkg/errors.TestErrorFormatting (0.00s)
PASS pkg/errors.TestErrorFind (0.00s)
PASS pkg/errors
EMPTY pkg/event
EMPTY pkg/handlers
EMPTY pkg/logger
PASS pkg/services.TestSQLTranslation (0.01s)
PASS pkg/services.TestValidateJsonbSearch/valid->>search (0.00s)
PASS pkg/services.TestValidateJsonbSearch/invalid_field_name_with_SQL_injection (0.00s)
PASS pkg/services.TestValidateJsonbSearch/invalid_name_with_SQL_injection (0.00s)
PASS pkg/services.TestValidateJsonbSearch/invalid_value (0.00s)
PASS pkg/services.TestValidateJsonbSearch/emtpty_value_is_valid (0.00s)
PASS pkg/services.TestValidateJsonbSearch/complex_labels (0.00s)
PASS pkg/services.TestValidateJsonbSearch/valid@>search (0.00s)
PASS pkg/services.TestValidateJsonbSearch/invalid_json_object,must_be_an_array (0.00s)
PASS pkg/services.TestValidateJsonbSearch/invalid_json_object,missed} (0.00s)
PASS pkg/services.TestValidateJsonbSearch/invalid_json_object_with_SQL_injection (0.00s)
PASS pkg/services.TestValidateJsonbSearch/invalid_search_without_field_name (0.00s)
PASS pkg/services.TestValidateJsonbSearch/invalid_search_with_two->>symbols (0.00s)
PASS pkg/services.TestValidateJsonbSearch (0.00s)
PASS pkg/services.TestResourceFindByConsumerID (0.00s)
PASS pkg/services.TestCreateInvalidResource (0.00s)
PASS pkg/services.TestValidateConsumer/validated (0.00s)
PASS pkg/services.TestValidateConsumer/wrong_name (0.00s)
PASS pkg/services.TestValidateConsumer/max_length (0.00s)
PASS pkg/services.TestValidateConsumer (0.00s)
PASS pkg/services.TestValidateResourceName/validated (0.00s)
PASS pkg/services.TestValidateResourceName/wrong_name (0.00s)
PASS pkg/services.TestValidateResourceName/max_length (0.00s)
PASS pkg/services.TestValidateResourceName (0.00s)
PASS pkg/services.TestValidateNewManifest/validated_single_manifest (0.00s)
PASS pkg/services.TestValidateNewManifest/validated_bundle_manifest (0.00s)
PASS pkg/services.TestValidateNewManifest/invalidated_single_manifest (0.00s)
PASS pkg/services.TestValidateNewManifest/invalidated_bundle_manifest (0.00s)
PASS pkg/services.TestValidateNewManifest/invalidated_resource_type (0.00s)
PASS pkg/services.TestValidateNewManifest (0.00s)
PASS pkg/services.TestValidateNewObject/validated (0.00s)
PASS pkg/services.TestValidateNewObject/no_apiVersion (0.00s)
PASS pkg/services.TestValidateNewObject/no_kind (0.00s)
PASS pkg/services.TestValidateNewObject/no_name (0.00s)
PASS pkg/services.TestValidateNewObject/has_generate_name (0.00s)
PASS pkg/services.TestValidateNewObject/has_resource_version (0.00s)
PASS pkg/services.TestValidateNewObject/has_deletion_grace_period_seconds (0.00s)
PASS pkg/services.TestValidateNewObject/wrong_namespace (0.00s)
PASS pkg/services.TestValidateNewObject/wrong_api_version(no_version) (0.00s)
PASS pkg/services.TestValidateNewObject/wrong_api_version(no_version)#1 (0.00s)
PASS pkg/services.TestValidateNewObject/wrong_api_version (0.00s)
PASS pkg/services.TestValidateNewObject (0.00s)
PASS pkg/services.TestValidateUpdateManifest/validated_single_manifest (0.00s)
PASS pkg/services.TestValidateUpdateManifest/validated_bundle_manifest (0.00s)
PASS pkg/services.TestValidateUpdateManifest/invalidated_single_manifest (0.00s)
PASS pkg/services.TestValidateUpdateManifest/invalidated_bundle_manifest (0.00s)
PASS pkg/services.TestValidateUpdateManifest/invalidated_resource_type (0.00s)
PASS pkg/services.TestValidateUpdateManifest (0.00s)
PASS pkg/services.TestValidateUpdateObject/validated (0.00s)
PASS pkg/services.TestValidateUpdateObject/apiVersion_mismatch (0.00s)
PASS pkg/services.TestValidateUpdateObject/kind_mismatch (0.00s)
PASS pkg/services.TestValidateUpdateObject/name_mismatch (0.00s)
PASS pkg/services.TestValidateUpdateObject/namespace_mismatch (0.00s)
PASS pkg/services.TestValidateUpdateObject (0.00s)
PASS pkg/services
EMPTY pkg/util
EMPTY cmd/maestro
EMPTY cmd/maestro/agent
PASS cmd/maestro/environments.TestLoadServices (0.27s)
PASS cmd/maestro/environments
EMPTY cmd/maestro/migrate
EMPTY cmd/maestro/servecmd
EMPTY cmd/maestro/server
EMPTY cmd/maestro/server/logging

DONE 84 tests in 31.515s

go clean -testcache

make test-integration
OCM_ENV=testing gotestsum --jsonfile-timing-events=/home/cloud-user/go/src/github.com/openshift-online/maestro/integration-test-results.json --format short-verbose -- -p 1 -ldflags -s -v -timeout 1h
./test/integration
I0619 23:01:14.225573 1773807 framework.go:75] Initializing testing environment
I0619 23:01:14.551158 1773807 framework.go:160] Using Mock OCM Authz Client
I0619 23:01:14.552831 1773807 framework.go:213] Disabling Sentry error reporting
I0619 23:01:14.558331 1773807 event.go:75] Starting event broadcaster
I0619 23:01:14.562721 1773807 grpc_server.go:73] Serving gRPC service without TLS at 8090
I0619 23:01:14.563110 1773807 healthcheck_server.go:56] Serving HealthCheck without TLS at 8083
{"level":"info","ts":1718852474.5633242,"logger":"fallback","caller":"[email protected]/protocol.go:124","msg":"subscribing to topics: map[sources/maestro/consumers/+/agentevents:{1 0 false false}]"}
I0619 23:01:14.563575 1773807 api_server.go:151] Serving without TLS at 8000
PASS test/integration.TestConsumerGet (0.07s)
PASS test/integration.TestConsumerPost (0.04s)
PASS test/integration.TestConsumerPatch (0.05s)
PASS test/integration.TestConsumerPaging (0.06s)
PASS test/integration.TestControllerRacing (1.34s)
PASS test/integration.TestControllerReconcile (1.14s)
PASS test/integration.TestControllerSync (1.04s)
PASS test/integration.TestPulseServer (7.04s)
PASS test/integration.TestResourceGet (0.05s)
PASS test/integration.TestResourcePost (2.07s)
PASS test/integration.TestResourcePostWithoutName (0.06s)
PASS test/integration.TestResourcePostWithName (0.06s)
PASS test/integration.TestResourcePatch (0.06s)
PASS test/integration.TestResourcePaging (0.20s)
PASS test/integration.TestResourceListSearch (0.10s)
PASS test/integration.TestResourceBundleGet (0.04s)
PASS test/integration.TestResourceBundleListSearch (0.19s)
PASS test/integration.TestUpdateResourceWithRacingRequests (0.13s)
PASS test/integration.TestResourceFromGRPC (5.09s)
PASS test/integration.TestResourceBundleFromGRPC (3.07s)
I0619 23:01:36.481128 1773807 api_server.go:157] Web server terminated
I0619 23:01:36.481457 1773807 healthcheck_server.go:60] HealthCheck server terminated
E0619 23:01:36.481755 1773807 baseclient.go:88] the cloudevents client is disconnected, read tcp 127.0.0.1:53096->127.0.0.1:1883: use of closed network connection
PASS test/integration

DONE 20 tests in 24.751s

@clyang82
Copy link
Contributor

/ok-to-test

@skeeey
Copy link
Contributor Author

skeeey commented Jun 20, 2024

/ok-to-test

@skeeey skeeey mentioned this pull request Jun 20, 2024
Closed
@machi1990
Copy link
Contributor

Thanks @skeeey

This specifically mentions when client is created. How about when a reconnect happens?

@skeeey skeeey force-pushed the sync branch 2 times, most recently from 52b815c to c19c4aa Compare June 21, 2024 01:33
@skeeey
Copy link
Contributor Author

skeeey commented Jun 21, 2024

@machi1990 thanks for reminding, we need handle the reconnect, I provided a NewMaestroGRPCSourceWorkClient instead of the sdk-go builder, in the NewMaestroGRPCSourceWorkClient, I also handled the reconnect case, so for the maestro work client, it will sync works from maestro server when

  1. a watcher is required
  2. the client reconnect happened, the client will sync works for its watchers

please take a look agian

@skeeey skeeey changed the title support full sync when the work watcher creating support full sync when the work watcher creating or the work client reconnected Jun 21, 2024
@skeeey skeeey force-pushed the sync branch 4 times, most recently from 47f200f to cec4ac1 Compare June 21, 2024 05:37
machi1990
machi1990 reviewed Jun 21, 2024
View reviewed changes
Copy link
Contributor

@machi1990 machi1990 left a comment •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks @skeeey

I've three works for the given source mw-client-example

maestro=# select id,source from resources;
                  id                  |      source       
--------------------------------------+-------------------
 80e73b12-017a-51ce-9b0e-0cde431aee03 | mw-client-example
 798a279a-fb9c-51d7-ab63-20b5c19ab294 | mw-client-example
 cb1a8c4b-544c-5ced-97a4-fa1ae882da42 | cs-example
 ab8e5a4a-aecc-5e20-9b43-fbe66c077d0a | mw-client-example
(4 rows)

However, when I run the watch on start I receive only one modified event change for the last work

:client-a/ (sync✗) $ go run main.go                                                                                                       [9:50:06]
{"level":"info","ts":1718956283.378795,"logger":"fallback","caller":"protocol/protocol.go:112","msg":"subscribing events for: mw-client-example"}
watched work (uid=ab8e5a4a-aecc-5e20-9b43-fbe66c077d0a) is modified

I suspect the queue is pop well ahead before the watching effectively starts on the client i.e the watcher has been registered;

maestro/pkg/client/cloudevents/grpcsource/watcherstore.go

Line 62 in cec4ac1

go wait.Until(s.process, time.Second, ctx.Done())
and

maestro/pkg/client/cloudevents/grpcsource/watcherstore.go

Line 270 in cec4ac1

m.sendWatchEvent(watch.Event{Type: watch.Modified, Object: work})
for references

@skeeey
Copy link
Contributor Author

skeeey commented Jun 21, 2024

the sendWatchEvent is a block operation, the queue pop should be blocked by this, I already test this, let me try this again

@machi1990
Copy link
Contributor

the sendWatchEvent is a block operation, the queue pop should be blocked by this, I already test this, let me try this again

As soon as the work is added

maestro/pkg/client/cloudevents/grpcsource/watcherstore.go

Line 92 in cec4ac1

if err := m.workQueue.Add(work); err != nil {
the process loop could start processing it even before the watcher registration has happened

maestro/pkg/client/cloudevents/grpcsource/watcherstore.go

Line 97 in cec4ac1

return m.registerWatcher(namespace), nil

I believe this is the main issue

@machi1990
Copy link
Contributor

lgtm

I created #140 to ensure that the client will work when jwt is enabled on Maestro's server. This doesn't block my testing though.

clyang82
clyang82 approved these changes Jun 21, 2024
View reviewed changes
Copy link
Contributor

@clyang82 clyang82 left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

LGTM

@clyang82 clyang82 merged commit d112d69 into openshift-online:main Jun 21, 2024
5 checks passed
@skeeey skeeey deleted the sync branch July 1, 2024 03:54
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Reviewers

@machi1990 machi1990 machi1990 left review comments

@clyang82 clyang82 clyang82 approved these changes

Assignees
No one assigned
Labels
None yet
Projects
None yet
Milestone
No milestone
Development

Successfully merging this pull request may close these issues.
3 participants
@skeeey @clyang82 @machi1990