Add CustomMessages demo examples (open-telemetry#496)

Added examples how to send and receive Custom Messages in example agent and server. 

This PR addresses open-telemetry#468. We improved examples while exploring custom messages.

Changes:
* in server UI there is new section where user see last xx received custom messages and can send custom message
* agent listens for specific custom message and send back response

Author: TomasVrabel

internal/examples/agent/agent/agent.go

@@ -9,6 +9,7 @@ import (
 	"crypto/x509"
 	"crypto/x509/pkix"
 	"encoding/pem"
+	"errors"
 	"fmt"
 	"net/http"
 	"os"
@@ -46,6 +47,8 @@ service:
       exporters: [otlp]
 `
 
+const customCapability_Health = "io.opentelemetry.custom.health"
+
 type Agent struct {
 	logger types.Logger
 
@@ -191,6 +194,16 @@ func (agent *Agent) connect(ops ...settingsOp) error {
 		return err
 	}
 
+	customCapabilities := &protobufs.CustomCapabilities{
+		Capabilities: []string{
+			customCapability_Health,
+		},
+	}
+	err = agent.opampClient.SetCustomCapabilities(customCapabilities)
+	if err != nil {
+		return err
+	}
+
 	// This sets the request to create a client certificate before the OpAMP client
 	// is started, before the connection is established. However, this assumes the
 	// server supports "AcceptsConnectionRequest" capability.
@@ -517,6 +530,10 @@ func (agent *Agent) onMessage(ctx context.Context, msg *types.MessageData) {
 		agent.updateAgentIdentity(ctx, uid)
 	}
 
+	if msg.CustomMessage != nil {
+		agent.processCustomMessage(ctx, msg.CustomMessage)
+	}
+
 	if configChanged {
 		err := agent.opampClient.UpdateEffectiveConfig(ctx)
 		if err != nil {
@@ -534,6 +551,53 @@ func (agent *Agent) onMessage(ctx context.Context, msg *types.MessageData) {
 	agent.requestClientCertificate()
 }
 
+func (agent *Agent) processCustomMessage(ctx context.Context, customMessage *protobufs.CustomMessage) {
+	if customMessage == nil {
+		return
+	}
+
+	agent.logger.Debugf(ctx, "received custom message: capability=%s, type=%s, data=%s",
+		customMessage.Capability,
+		customMessage.Type,
+		string(customMessage.Data),
+	)
+
+	if customMessage.Capability == customCapability_Health {
+		switch customMessage.Type {
+		case "get_health":
+
+			responseMsg := &protobufs.CustomMessage{
+				Capability: customCapability_Health,
+				Type:       "health_status",
+				Data:       []byte("OK"),
+			}
+
+			if err := agent.sendCustomMessage(ctx, responseMsg); err != nil {
+				agent.logger.Errorf(ctx, "Failed to send custom message: %v", err)
+			}
+		default:
+			agent.logger.Errorf(ctx, "unknown custom message type: %s", customMessage.Type)
+		}
+	}
+}
+
+func (agent *Agent) sendCustomMessage(ctx context.Context, message *protobufs.CustomMessage) error {
+	for {
+		sendingChan, err := agent.opampClient.SendCustomMessage(message)
+
+		switch {
+		case err == nil:
+			return nil
+		case errors.Is(err, types.ErrCustomMessagePending):
+			// Failed to send opamp message. Waiting for previous custom message to be sent
+			<-sendingChan
+		default:
+			agent.logger.Errorf(ctx, "failed to send custom message: %w", err)
+			return fmt.Errorf("failed to send custom message: %w", err)
+		}
+	}
+}
+
 func (agent *Agent) tryChangeOpAMP(ctx context.Context, cert *tls.Certificate, tlsConfig *tls.Config, proxy *proxySettings) {
 	agent.logger.Debugf(ctx, "Reconnecting to verify new OpAMP settings.\n")
 	agent.disconnect(ctx)

internal/examples/html/html/agent.html

@@ -147,3 +147,34 @@ <h3>OpAMP Connection Settings</h3>
 
 </body>
 </html>
+
+<hr/>
+
+<h3>Custom Messages</h3>
+
+<h4>Send Custom Message</h4>
+<form action="/send_custom_message" method="post">
+    <input type="hidden" name="instanceid" value="{{ .InstanceIdStr }}"/>
+    
+    <label for="capability">Capability:</label>
+        <input type="text" name="capability" value="io.opentelemetry.custom.health" size="30"/><br/>
+
+    <label for="type">Type:</label>
+        <input type="text" name="type" value="get_health" size="30"/><br/>
+
+    <label for="data">Data:</label>
+        <input type="text" name="data" value="" size="30"/><br/>
+
+    <input type="submit" value="Send Custom Message" />
+</form>
+
+<h4>Received Messages</h4>
+
+{{ if eq (len .CustomMessageHistory) 0 }}
+No Received Messages
+{{ else }}
+Last {{ len .CustomMessageHistory }} Received Messages
+<pre class="log-container"><code>{{ range .CustomMessageHistory }}{{ . }}
+{{ end }}
+</code></pre>
+{{ end }} 

internal/examples/server/data/agent.go

@@ -19,6 +19,10 @@ import (
 	"github.com/open-telemetry/opamp-go/server/types"
 )
 
+const (
+	CustomMessageHistorySize = 15
+)
+
 // Agent represents a connected Agent.
 type Agent struct {
 	// Some fields in this struct are exported so that we can render them in the UI.
@@ -51,6 +55,9 @@ type Agent struct {
 	ClientCertSha256Fingerprint string
 	ClientCertOfferError        string
 
+	// Custom message history
+	CustomMessageHistory []string
+
 	// Remote config that we will give to this Agent.
 	remoteConfig *protobufs.AgentRemoteConfig
 
@@ -95,6 +102,7 @@ func (agent *Agent) CloneReadonly() *Agent {
 		ClientCert:                  agent.ClientCert,
 		ClientCertOfferError:        agent.ClientCertOfferError,
 		ClientCertSha256Fingerprint: agent.ClientCertSha256Fingerprint,
+		CustomMessageHistory:        agent.CustomMessageHistory,
 	}
 }
 
@@ -113,6 +121,10 @@ func (agent *Agent) UpdateStatus(
 		agent.processConnectionSettingsRequest(statusMsg.ConnectionSettingsRequest.Opamp, response)
 	}
 
+	if statusMsg.CustomMessage != nil {
+		agent.processCustomMessage(statusMsg)
+	}
+
 	statusUpdateWatchers := agent.statusUpdateWatchers
 	agent.statusUpdateWatchers = nil
 
@@ -122,6 +134,28 @@ func (agent *Agent) UpdateStatus(
 	notifyStatusWatchers(statusUpdateWatchers)
 }
 
+func (agent *Agent) processCustomMessage(statusMsg *protobufs.AgentToServer) {
+	formattedMessage := fmt.Sprintf("[%s] capability=%s, type=%s, data=%s",
+		time.Now().Format(time.DateTime),
+		statusMsg.CustomMessage.Capability,
+		statusMsg.CustomMessage.Type,
+		string(statusMsg.CustomMessage.Data),
+	)
+
+	agent.CustomMessageHistory = append(agent.CustomMessageHistory, formattedMessage)
+
+	if len(agent.CustomMessageHistory) > CustomMessageHistorySize {
+		agent.CustomMessageHistory = agent.CustomMessageHistory[len(agent.CustomMessageHistory)-CustomMessageHistorySize:]
+	}
+}
+
+func (agent *Agent) SendCustomMessage(customMsg *protobufs.ServerToAgent) {
+	agent.mux.Lock()
+	defer agent.mux.Unlock()
+
+	agent.SendToAgent(customMsg)
+}
+
 func notifyStatusWatchers(statusUpdateWatchers []chan<- struct{}) {
 	// Notify everyone who is waiting on this Agent's status updates.
 	for _, ch := range statusUpdateWatchers {

internal/examples/server/data/agents.go

@@ -31,6 +31,16 @@ func (agents *Agents) RemoveConnection(conn types.Connection) {
 	delete(agents.connections, conn)
 }
 
+func (agents *Agents) SendCustomMessageToAgent(
+	agentId InstanceId,
+	customMsg *protobufs.ServerToAgent,
+) {
+	agent := agents.FindAgent(agentId)
+	if agent != nil {
+		agent.SendCustomMessage(customMsg)
+	}
+}
+
 func (agents *Agents) SetCustomConfigForAgent(
 	agentId InstanceId,
 	config *protobufs.AgentConfigMap,

internal/examples/server/uisrv/ui.go

@@ -33,6 +33,7 @@ func Start(rootDir string) {
 	mux.HandleFunc("/save_config", saveCustomConfigForInstance)
 	mux.HandleFunc("/rotate_client_cert", rotateInstanceClientCert)
 	mux.HandleFunc("/opamp_connection_settings", opampConnectionSettings)
+	mux.HandleFunc("/send_custom_message", sendCustomMessage)
 	srv = &http.Server{
 		Addr:    "0.0.0.0:4321",
 		Handler: mux,
@@ -80,6 +81,42 @@ func renderAgent(w http.ResponseWriter, r *http.Request) {
 	renderTemplate(w, "agent.html", agent)
 }
 
+func sendCustomMessage(w http.ResponseWriter, r *http.Request) {
+	if err := r.ParseForm(); err != nil {
+		w.WriteHeader(http.StatusInternalServerError)
+		return
+	}
+
+	uid, err := uuid.Parse(r.Form.Get("instanceid"))
+	if err != nil {
+		w.WriteHeader(http.StatusNotFound)
+		return
+	}
+
+	instanceId := data.InstanceId(uid)
+	agent := data.AllAgents.GetAgentReadonlyClone(instanceId)
+	if agent == nil {
+		w.WriteHeader(http.StatusNotFound)
+		return
+	}
+
+	capability := r.PostForm.Get("capability")
+	msgType := r.PostForm.Get("type")
+	dataStr := r.PostForm.Get("data")
+
+	customMsg := &protobufs.ServerToAgent{
+		CustomMessage: &protobufs.CustomMessage{
+			Capability: capability,
+			Type:       msgType,
+			Data:       []byte(dataStr),
+		},
+	}
+
+	data.AllAgents.SendCustomMessageToAgent(instanceId, customMsg)
+
+	http.Redirect(w, r, "/agent?instanceid="+uid.String(), http.StatusSeeOther)
+}
+
 func saveCustomConfigForInstance(w http.ResponseWriter, r *http.Request) {
 	if err := r.ParseForm(); err != nil {
 		w.WriteHeader(http.StatusInternalServerError)