Skip to content

Commit

Permalink
Merge branch 'openshift-online:main' into main
Browse files Browse the repository at this point in the history
  • Loading branch information
clyang82 authored May 17, 2024
2 parents f3cc2e9 + 8ad292b commit 8e3e30d
Show file tree
Hide file tree
Showing 39 changed files with 905 additions and 184 deletions.
36 changes: 36 additions & 0 deletions .github/workflows/e2e.yml
Original file line number Diff line number Diff line change
@@ -0,0 +1,36 @@
name: E2E Test

on:
workflow_dispatch: {}
pull_request:
branches:
- main

env:
GO_VERSION: '1.21'
GO_REQUIRED_MIN_VERSION: ''

permissions:
contents: read

jobs:
e2e:
runs-on: ubuntu-latest
steps:
- name: Checkout
uses: actions/checkout@v4
- name: Setup Go
uses: actions/setup-go@v5
with:
go-version: ${{ env.GO_VERSION }}
- name: Setup kind
uses: engineerd/[email protected]
with:
version: v0.17.0
- name: install ginkgo
run: go install github.com/onsi/ginkgo/v2/[email protected]
- name: Test E2E
run: |
make e2e-test
env:
container_tool: docker
1 change: 0 additions & 1 deletion Dockerfile
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,6 @@ COPY . $SOURCE_DIR

ENV GOFLAGS=""
RUN make binary
RUN pwd

FROM registry.access.redhat.com/ubi9/ubi-minimal:latest

Expand Down
4 changes: 2 additions & 2 deletions Makefile
Original file line number Diff line number Diff line change
Expand Up @@ -195,7 +195,7 @@ install: check-gopath
# Examples:
# make test TESTFLAGS="-run TestSomething"
test:
OCM_ENV=testing gotestsum --jsonfile-timing-events=$(unit_test_json_output) --format short-verbose -- -p 1 -v $(TESTFLAGS) \
OCM_ENV=testing gotestsum --jsonfile-timing-events=$(unit_test_json_output) --format $(TEST_SUMMARY_FORMAT) -- -p 1 -v $(TESTFLAGS) \
./pkg/... \
./cmd/...
.PHONY: test
Expand Down Expand Up @@ -268,7 +268,7 @@ cmds:
--param="DATABASE_PASSWORD=$(db_password)" \
--param="DATABASE_PORT=$(db_port)" \
--param="DATABASE_USER=$(db_user)" \
--param="DATABASE_SSLMODE=$(db_sslmode)" \
--param="DB_SSLMODE=$(db_sslmode)" \
--param="POSTGRES_IMAGE=$(POSTGRES_IMAGE)" \
--param="MQTT_HOST=$(mqtt_host)" \
--param="MQTT_PORT=$(mqtt_port)" \
Expand Down
39 changes: 36 additions & 3 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -125,6 +125,16 @@ ocm get /api/maestro/v1/resources
}
```

#### Create a consumer:

```
ocm post /api/maestro/v1/consumers << EOF
{
"name": "cluster1"
}
EOF
```

#### Post a new Resource

```shell
Expand Down Expand Up @@ -402,8 +412,31 @@ $ oc get pod -n default
NAME READY STATUS RESTARTS AGE
nginx-5d6b548959-829c7 1/1 Running 0 70s
```

### Make a new Kind
## Make a new Kind

1. Add to openapi.yaml
2. Generate the new structs/clients (`make generate`)
2. Generate the new structs/clients (`make generate`)

## Configure maestro server

### MQTT Configuration

Using the `--mqtt-config-file` to specify the MQTT configuration file for maestro server, the format of the configuration file can be yaml or json, it contains the following configurations

```yaml
brokerHost: <MQTT broker host, e.g. 127.0.0.1:1883>
username: <the username for MQTT broker, if required by username and password authentication>
password: <the password for MQTT broker, if required by username and password authentication>
caFile: <the CA of the MQTT broker, if required by mTLS authentication>
clientCertFile: <the cert of the MQTT client, if required by mTLS authentication>
clientKeyFile: <the cert key of the MQTT client, if required by mTLS authentication>
topics:
sourceEvents: sources/maestro/consumers/+/sourceevents
agentEvents: <the topic for agent events>
```
For `topics.agentEvents`

- If the MQTT broker supports the [shared subscriptions](
https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901250), the topic needs to be set to `$share/statussubscribers/sources/maestro/consumers/+/agentevents`
- If the MQTT broker does not support the shared subscriptions, the topic needs to be set to `sources/maestro/consumers/+/agentevents` and set the maestro server flag `--subscription-type` to `broadcast`
17 changes: 9 additions & 8 deletions cmd/maestro/agent/cmd.go
Original file line number Diff line number Diff line change
Expand Up @@ -67,9 +67,6 @@ func runAgent(cmd *cobra.Command, args []string) {

// use mqtt as the default driver
agentOption.MaxJSONRawLength = maxJSONRawLength
agentOption.WorkloadSourceDriver = "mqtt"
agentOption.CloudEventsClientCodecs = []string{"manifest"}

cfg := spoke.NewWorkAgentConfig(commonOptions, agentOption)
cmdConfig := commonOptions.CommoOpts.
NewControllerCommandConfig("maestro-agent", version.Get(), cfg.RunWorkloadAgent)
Expand All @@ -90,9 +87,13 @@ func addFlags(fs *pflag.FlagSet) {
agentOption.AppliedManifestWorkEvictionGracePeriod, "Grace period for resource eviction")
fs.StringVar(&commonOptions.SpokeClusterName, "consumer-name",
commonOptions.SpokeClusterName, "Name of the consumer")
// mqtt config file
fs.StringVar(&agentOption.WorkloadSourceConfig, "mqtt-config-file",
agentOption.WorkloadSourceConfig, "The config file path of mqtt broker")
fs.StringVar(&agentOption.CloudEventsClientID, "mqtt-client-id",
agentOption.CloudEventsClientID, "The ID of the mqtt client, by default it is <consumer-id>-work-agent")
// message broker config file
fs.StringVar(&agentOption.WorkloadSourceConfig, "message-broker-config-file",
agentOption.WorkloadSourceConfig, "The config file path of the message broker, it can be mqtt broker or kafka broker")
fs.StringVar(&agentOption.WorkloadSourceDriver, "message-broker-type", "mqtt", "Message broker type (default: mqtt)")
fs.StringVar(&agentOption.CloudEventsClientID, "agent-client-id",
agentOption.CloudEventsClientID, "The ID of the agent client, by default it is <consumer-id>-work-agent")
fs.StringSliceVar(&agentOption.CloudEventsClientCodecs, "agent-client-codecs",
[]string{"manifest"}, "The codecs of the agent client. The valid codecs are manifest and manifestbundle")

}
26 changes: 20 additions & 6 deletions cmd/maestro/environments/framework.go
Original file line number Diff line number Diff line change
Expand Up @@ -7,14 +7,13 @@ import (

"github.com/getsentry/sentry-go"
"github.com/golang/glog"
"github.com/spf13/pflag"

"github.com/openshift-online/maestro/pkg/client/cloudevents"
"github.com/openshift-online/maestro/pkg/client/ocm"
"github.com/openshift-online/maestro/pkg/config"
"github.com/openshift-online/maestro/pkg/errors"
"github.com/spf13/pflag"

mqttoptions "open-cluster-management.io/sdk-go/pkg/cloudevents/generic/options/mqtt"
"open-cluster-management.io/sdk-go/pkg/cloudevents/generic"
)

func init() {
Expand Down Expand Up @@ -173,12 +172,27 @@ func (e *Env) LoadClients() error {
glog.Infof("Using Mock CloudEvents Source Client")
e.Clients.CloudEventsSource = cloudevents.NewSourceClientMock(e.Services.Resources())
} else {
cloudEventsSourceOptions := mqttoptions.NewSourceOptions(e.Config.MessageBroker.MQTTOptions, e.Config.MessageBroker.ClientID, e.Config.MessageBroker.SourceID)
e.Clients.CloudEventsSource, err = cloudevents.NewSourceClient(cloudEventsSourceOptions, e.Services.Resources())

_, config, err := generic.NewConfigLoader(e.Config.MessageBroker.MessageBrokerType, e.Config.MessageBroker.MessageBrokerConfig).
LoadConfig()
if err != nil {
glog.Errorf("Unable to create CloudEvents Source client: %s", err.Error())
glog.Errorf("Unable to load configuration: %s", err.Error())
return err
}

cloudEventsSourceOptions, err := generic.BuildCloudEventsSourceOptions(config,
e.Config.MessageBroker.ClientID, e.Config.MessageBroker.SourceID)
if err != nil {
glog.Errorf("Unable to build cloudevent source options: %s", err.Error())
return err
}
if cloudEventsSourceOptions != nil {
e.Clients.CloudEventsSource, err = cloudevents.NewSourceClient(cloudEventsSourceOptions, e.Services.Resources())
if err != nil {
glog.Errorf("Unable to create CloudEvents Source client: %s", err.Error())
return err
}
}
}

return nil
Expand Down
12 changes: 7 additions & 5 deletions cmd/maestro/server/grpc_server.go
Original file line number Diff line number Diff line change
Expand Up @@ -107,6 +107,8 @@ func (svr *GRPCServer) Publish(ctx context.Context, pubReq *pbv1.PublishRequest)
return nil, fmt.Errorf("failed to parse cloud event type %s, %v", evt.Type(), err)
}

glog.V(4).Infof("receive the event with grpc, %s", evt)

// handler resync request
if eventType.Action == types.ResyncRequestAction {
err := svr.respondResyncStatusRequest(ctx, eventType.CloudEventsDataType, evt)
Expand All @@ -133,11 +135,9 @@ func (svr *GRPCServer) Publish(ctx context.Context, pubReq *pbv1.PublishRequest)
if err != nil {
return nil, fmt.Errorf("failed to get resource: %v", err)
}
// handle the special case that the resource is updated by the source controller
// and the version of the resource in the request is less than it in the database
if found.Version < res.Version {
res.Version = found.Version
}
// keep the existing version for bundle resource, mainly from hub controller,
// the version is not guaranteed to be increased.
res.Version = found.Version
}
_, err := svr.resourceService.Update(ctx, res)
if err != nil {
Expand All @@ -163,6 +163,8 @@ func (svr *GRPCServer) Subscribe(subReq *pbv1.SubscriptionRequest, subServer pbv
return fmt.Errorf("failed to encode resource %s to cloudevent: %v", res.ID, err)
}

glog.V(4).Infof("send the event with grpc, %s", evt)

// WARNING: don't use "pbEvt, err := pb.ToProto(evt)" to convert cloudevent to protobuf
pbEvt := &pbv1.CloudEvent{}
if err = grpcprotocol.WritePBMessage(context.TODO(), binding.ToMessage(evt), pbEvt); err != nil {
Expand Down
42 changes: 32 additions & 10 deletions cmd/maestro/server/pulse_server.go
Original file line number Diff line number Diff line change
Expand Up @@ -151,15 +151,29 @@ func (s *PulseServer) startSubscription(ctx context.Context) {
log.V(1).Infof("received action %s for resource %s", action, resource.ID)
switch action {
case types.StatusModified:
found, svcErr := s.resourceService.Get(ctx, resource.ID)
if svcErr != nil {
if svcErr.Is404() {
log.Warning(fmt.Sprintf("skipping resource %s as it is not found", resource.ID))
return nil
}

return fmt.Errorf("failed to get resource %s, %s", resource.ID, svcErr.Error())
}

if found.ConsumerName != resource.ConsumerName {
return fmt.Errorf("unmatched consumer name %s for resource %s", resource.ConsumerName, resource.ID)
}

// set the resource source back for broadcast
resource.Source = found.Source

if !s.statusDispatcher.Dispatch(resource.ConsumerName) {
// the resource is not owned by the current instance, skip
log.V(4).Infof("skipping resource status update %s as it is not owned by the current instance", resource.ID)
return nil
}

// broadcast the resource status update event
s.eventBroadcaster.Broadcast(resource)

// convert the resource status to cloudevent
evt, err := api.JSONMAPToCloudEvent(resource.Status)
if err != nil {
Expand All @@ -174,14 +188,22 @@ func (s *PulseServer) startSubscription(ctx context.Context) {

// if the resource has been deleted from agent, delete it from maestro
if meta.IsStatusConditionTrue(statusPayload.Conditions, common.ManifestsDeleted) {
if err := s.resourceService.Delete(ctx, resource.ID); err != nil {
return err
}
} else {
// update the resource status
if _, err := s.resourceService.UpdateStatus(ctx, resource); err != nil {
return err
if svcErr := s.resourceService.Delete(ctx, resource.ID); svcErr != nil {
return svcErr
}

s.eventBroadcaster.Broadcast(resource)
return nil
}
// update the resource status
_, updated, svcErr := s.resourceService.UpdateStatus(ctx, resource)
if svcErr != nil {
return svcErr
}

// broadcast the resource status updated only when the resource is updated
if updated {
s.eventBroadcaster.Broadcast(resource)
}
default:
return fmt.Errorf("unsupported action %s", action)
Expand Down
4 changes: 2 additions & 2 deletions data/generated/openapi/openapi.go

Large diffs are not rendered by default.

46 changes: 24 additions & 22 deletions go.mod
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@ require (
github.com/bwmarrin/snowflake v0.3.0
github.com/bxcodec/faker/v3 v3.2.0
github.com/cespare/xxhash v1.1.0
github.com/cloudevents/sdk-go/v2 v2.14.0
github.com/cloudevents/sdk-go/v2 v2.15.3-0.20240329120647-e6a74efbacbf
github.com/deckarep/golang-set/v2 v2.6.0
github.com/docker/go-healthcheck v0.1.0
github.com/getsentry/sentry-go v0.20.0
Expand All @@ -23,8 +23,8 @@ require (
github.com/jinzhu/inflection v1.0.0
github.com/lib/pq v1.10.7
github.com/mendsley/gojwk v0.0.0-20141217222730-4d5ec6e58103
github.com/onsi/ginkgo/v2 v2.15.0
github.com/onsi/gomega v1.31.1
github.com/onsi/ginkgo/v2 v2.17.1
github.com/onsi/gomega v1.32.0
github.com/openshift-online/ocm-sdk-go v0.1.334
github.com/prometheus/client_golang v1.18.0
github.com/segmentio/ksuid v1.0.2
Expand All @@ -37,13 +37,13 @@ require (
gorm.io/datatypes v1.2.0
gorm.io/driver/postgres v1.5.0
gorm.io/gorm v1.24.7-0.20230306060331-85eaf9eeda11
k8s.io/apimachinery v0.29.2
k8s.io/client-go v0.29.2
k8s.io/component-base v0.29.1
k8s.io/apimachinery v0.29.4
k8s.io/client-go v0.29.4
k8s.io/component-base v0.29.3
k8s.io/klog/v2 v2.120.1
open-cluster-management.io/api v0.13.0
open-cluster-management.io/ocm v0.13.1-0.20240313094829-1c0c0156e780
open-cluster-management.io/sdk-go v0.13.1-0.20240321032811-7dbdd1b5c63d
open-cluster-management.io/api v0.13.1-0.20240506072237-800b00d9f0db
open-cluster-management.io/ocm v0.13.1-0.20240514020334-4117a4b3027f
open-cluster-management.io/sdk-go v0.13.1-0.20240516092635-a00a7ab51fd2
)

require (
Expand All @@ -56,15 +56,17 @@ require (
github.com/blang/semver/v4 v4.0.0 // indirect
github.com/cenkalti/backoff/v4 v4.2.1 // indirect
github.com/cespare/xxhash/v2 v2.2.0 // indirect
github.com/cloudevents/sdk-go/protocol/kafka_confluent/v2 v2.0.0-20240413090539-7fef29478991 // indirect
github.com/cloudevents/sdk-go/protocol/mqtt_paho/v2 v2.0.0-20231030012137-0836a524e995 // indirect
github.com/confluentinc/confluent-kafka-go/v2 v2.3.0 // indirect
github.com/coreos/go-semver v0.3.1 // indirect
github.com/coreos/go-systemd/v22 v22.5.0 // indirect
github.com/davecgh/go-spew v1.1.1 // indirect
github.com/dgrijalva/jwt-go v3.2.0+incompatible // indirect
github.com/docker/distribution v2.8.1+incompatible // indirect
github.com/eclipse/paho.golang v0.11.0 // indirect
github.com/emicklei/go-restful/v3 v3.11.0 // indirect
github.com/evanphx/json-patch v5.7.0+incompatible // indirect
github.com/evanphx/json-patch v5.9.0+incompatible // indirect
github.com/felixge/httpsnoop v1.0.3 // indirect
github.com/fsnotify/fsnotify v1.7.0 // indirect
github.com/go-logr/logr v1.4.1 // indirect
Expand Down Expand Up @@ -127,17 +129,17 @@ require (
go.opentelemetry.io/otel/trace v1.19.0 // indirect
go.opentelemetry.io/proto/otlp v1.0.0 // indirect
go.uber.org/multierr v1.11.0 // indirect
go.uber.org/zap v1.26.0 // indirect
golang.org/x/crypto v0.19.0 // indirect
go.uber.org/zap v1.27.0 // indirect
golang.org/x/crypto v0.21.0 // indirect
golang.org/x/exp v0.0.0-20230522175609-2e198f4a06a1 // indirect
golang.org/x/net v0.21.0 // indirect
golang.org/x/net v0.23.0 // indirect
golang.org/x/oauth2 v0.16.0 // indirect
golang.org/x/sync v0.6.0 // indirect
golang.org/x/sys v0.17.0 // indirect
golang.org/x/term v0.17.0 // indirect
golang.org/x/sys v0.18.0 // indirect
golang.org/x/term v0.18.0 // indirect
golang.org/x/text v0.14.0 // indirect
golang.org/x/time v0.3.0 // indirect
golang.org/x/tools v0.16.1 // indirect
golang.org/x/tools v0.17.0 // indirect
google.golang.org/appengine v1.6.8 // indirect
google.golang.org/genproto v0.0.0-20240123012728-ef4313101c80 // indirect
google.golang.org/genproto/googleapis/api v0.0.0-20240123012728-ef4313101c80 // indirect
Expand All @@ -147,15 +149,15 @@ require (
gopkg.in/yaml.v2 v2.4.0 // indirect
gopkg.in/yaml.v3 v3.0.1 // indirect
gorm.io/driver/mysql v1.4.7 // indirect
k8s.io/api v0.29.2 // indirect
k8s.io/apiextensions-apiserver v0.29.0 // indirect
k8s.io/apiserver v0.29.0 // indirect
k8s.io/kms v0.29.0 // indirect
k8s.io/kube-aggregator v0.29.0 // indirect
k8s.io/api v0.29.4 // indirect
k8s.io/apiextensions-apiserver v0.29.3 // indirect
k8s.io/apiserver v0.29.3 // indirect
k8s.io/kms v0.29.3 // indirect
k8s.io/kube-aggregator v0.29.3 // indirect
k8s.io/kube-openapi v0.0.0-20231010175941-2dd684a91f00 // indirect
k8s.io/utils v0.0.0-20240310230437-4693a0247e57 // indirect
sigs.k8s.io/apiserver-network-proxy/konnectivity-client v0.28.0 // indirect
sigs.k8s.io/controller-runtime v0.17.2 // indirect
sigs.k8s.io/controller-runtime v0.17.3 // indirect
sigs.k8s.io/json v0.0.0-20221116044647-bc3834ca7abd // indirect
sigs.k8s.io/kube-storage-version-migrator v0.0.6-0.20230721195810-5c8923c5ff96 // indirect
sigs.k8s.io/structured-merge-diff/v4 v4.4.1 // indirect
Expand Down
Loading

0 comments on commit 8e3e30d

Please sign in to comment.