Golang与Mesosphere Marathon PAAS交互的插件库go-marathon的使用

概述

go-marathon是一个用于与Marathon交互的API库，支持以下功能：

• 应用程序和组部署
• 用于获取状态、配置和任务的辅助过滤器
• 支持HA部署的多个端点
• Marathon事件订阅和事件流
• Pods

注意：该库仍在积极开发中，用户应预期API可能会有频繁变化（包括破坏性变更）。要求Go 1.6或更高版本。

代码示例

创建客户端

import (
	marathon "github.com/gambol99/go-marathon"
)

marathonURL := "http://10.241.1.71:8080"
config := marathon.NewDefaultConfig()
config.URL = marathonURL
client, err := marathon.NewClient(config)
if err != nil {
	log.Fatalf("Failed to create a client for marathon, error: %s", err)
}

applications, err := client.Applications(nil)
...

可以指定多个Marathon端点（例如在HA模式下运行Marathon时）：

marathonURL := "http://10.241.1.71:8080,10.241.1.72:8080,10.241.1.73:8080"

自定义HTTP客户端

可以传递自定义HTTP客户端以自定义行为（例如绕过TLS验证、加载根CA或更改超时）：

marathonURL := "http://10.241.1.71:8080"
config := marathon.NewDefaultConfig()
config.URL = marathonURL
config.HTTPClient = &http.Client{
    Timeout: (time.Duration(10) * time.Second),
    Transport: &http.Transport{
        Dial: (&net.Dialer{
            Timeout:   10 * time.Second,
            KeepAlive: 10 * time.Second,
        }).Dial,
        TLSClientConfig: &tls.Config{
            InsecureSkipVerify: true,
        },
    },
}
config.HTTPSSEClient = &http.Client{
    // 不能设置Timeout，因为它包含读取响应体的超时
    Transport: &http.Transport{
        Dial: (&net.Dialer{
            Timeout:   10 * time.Second,
            KeepAlive: 10 * time.Second,
        }).Dial,
        TLSClientConfig: &tls.Config{
            InsecureSkipVerify: true,
        },
    },
}

列出应用程序

applications, err := client.Applications(nil)
if err != nil {
	log.Fatalf("Failed to list applications: %s", err)
}

log.Printf("Found %d application(s) running", len(applications.Apps))
for _, application := range applications.Apps {
	log.Printf("Application: %s", application)
	appID := application.ID

	details, err := client.Application(appID)
	if err != nil {
		log.Fatalf("Failed to get application %s: %s", appID, err)
	}
	if details.Tasks != nil {
		for _, task := range details.Tasks {
			log.Printf("application %s has task: %s", appID, task)
		}
	}
}

创建新应用程序

log.Printf("Deploying a new application")
application := marathon.NewDockerApplication().
  Name(applicationName).
  CPU(0.1).
  Memory(64).
  Storage(0.0).
  Count(2).
  AddArgs("/usr/sbin/apache2ctl", "-D", "FOREGROUND").
  AddEnv("NAME", "frontend_http").
  AddEnv("SERVICE_80_NAME", "test_http").
  CheckHTTP("/health", 10, 5)

application.
  Container.Docker.Container("quay.io/gambol99/apache-php:latest").
  Bridged().
  Expose(80).
  Expose(443)

if _, err := client.CreateApplication(application); err != nil {
	log.Fatalf("Failed to create application: %s, error: %s", application, err)
} else {
	log.Printf("Created the application: %s", application)
}

扩展应用程序

将应用程序实例数更改为4：

log.Printf("Scale to 4 instances")
if err := client.ScaleApplicationInstances(application.ID, 4); err != nil {
	log.Fatalf("Failed to delete the application: %s, error: %s", application, err)
} else {
	client.WaitOnApplication(application.ID, 30 * time.Second)
	log.Printf("Successfully scaled the application")
}

Pods

Pods允许您将任务组作为单元部署。Pod中所有任务共享网络和存储：

// 初始化运行nginx的单容器pod
pod := marathon.NewPod()

image := marathon.NewDockerPodContainerImage().SetID("nginx")

container := marathon.NewPodContainer().
	SetName("container", i).
	CPUs(0.1).
	Memory(128).
	SetImage(image)

pod.Name("mypod").AddContainer(container)

// 创建并等待启动
pod, err := client.CreatePod(pod)
err = client.WaitOnPod(pod.ID, time.Minute*1)

// 扩展
pod.Count(5)
pod, err = client.UpdatePod(pod, true)

// 删除
id, err := client.DeletePod(pod.ID, true)

订阅和事件

事件流

仅适用于Marathon >= 0.9.0，不需要特殊配置：

// 配置客户端
config := marathon.NewDefaultConfig()
config.URL = marathonURL
config.EventsTransport = marathon.EventsTransportSSE

client, err := marathon.NewClient(config)
if err != nil {
	log.Fatalf("Failed to create a client for marathon, error: %s", err)
}

// 注册事件监听
events, err = client.AddEventsListener(marathon.EventIDApplications)
if err != nil {
	log.Fatalf("Failed to register for events, %s", err)
}

timer := time.After(60 * time.Second)
done := false

// 从通道接收事件60秒
for {
	if done {
		break
	}
	select {
	case <-timer:
		log.Printf("Exiting the loop")
		done = true
	case event := <-events:
		log.Printf("Received event: %s", event)
	}
}

// 取消订阅Marathon事件
client.RemoveEventsListener(events)

事件订阅

需要启动内置Web服务器：

// 配置客户端
config := marathon.NewDefaultConfig()
config.URL = marathonURL
config.EventsInterface = marathonInterface
config.EventsPort = marathonPort

client, err := marathon.NewClient(config)
if err != nil {
	log.Fatalf("Failed to create a client for marathon, error: %s", err)
}

// 注册事件监听
events, err = client.AddEventsListener(marathon.EventIDApplications)
if err != nil {
	log.Fatalf("Failed to register for events, %s", err)
}

timer := time.After(60 * time.Second)
done := false

// 从通道接收事件60秒
for {
	if done {
		break
	}
	select {
	case <-timer:
		log.Printf("Exiting the loop")
		done = true
	case event := <-events:
		log.Printf("Received event: %s", event)
	}
}

// 取消订阅Marathon事件
client.RemoveEventsListener(events)

控制订阅

// 配置客户端
config := marathon.NewDefaultConfig()
config.URL = marathonURL

client, err := marathon.NewClient(config)
if err != nil {
	log.Fatalf("Failed to create a client for marathon, error: %s", err)
}

// 通过回调URL注册事件订阅者
callbackURL := "http://10.241.1.71:9494"
if err := client.Subscribe(callbackURL); err != nil {
	log.Fatalf("Unable to register the callbackURL [%s], error: %s", callbackURL, err)
}

// 注销同一订阅者
if err := client.Unsubscribe(callbackURL); err != nil {
	log.Fatalf("Unable to deregister the callbackURL [%s], error: %s", callbackURL, err)
}