最近公司将我们之前使用的链路工具切换为了 OpenTelemetry
.
我们的技术栈是:
OTLP
Client──────────►Collect────────►StartRocks
(Agent) ▲
│
│
Jaeger
其中客户端使用 OpenTelemetry 提供的 Java Agent 进行埋点收集数据,再由 Agent 通过 OTLP(OpenTelemetry Protocol) 协议将数据发往 Collector,在 Collector
中我们可以自行任意处理数据,并决定将这些数据如何存储(这点在以往的 SkyWalking 体系中是很难自定义的)
这里我们将数据写入 StartRocks 中,供之后的 UI 层进行查看。
OpenTelemetry
是可观测系统的新标准,基于它可以兼容以前使用的 Prometheus、 victoriametrics、skywalking 等系统,同时还可以灵活扩展,不用与任何但一生态或技术栈进行绑定。更多关于 OTel 的内容会在今后介绍。
其中有一个关键问题就是:如何在线上进行无缝切换。
虽然我们内部的发布系统已经支持重新发布后就会切换到新的链路,也可以让业务自行发布然后逐步的切换到新的系统,这样也是最保险的方式。
但这样会有几个问题:
所以最好的方式还是由我们在后台统一发布,对外没有任何感知就可以一键全部切换为 OpenTelemetry。
仔细一看貌似也没什么难的,无非就是模拟用户点击发布按钮而已。
但这事由我们自动来做就不一样了,用户点击发布的时候会选择他们认为可以发布的分支进行发布,我们不能自作主张的比如选择 main 分支,有可能只是合并了但还不具备发布条件。
所以保险的方式还是得用当前项目上一次发布时所使用的 git hash 值重新打包发布。
但这也有几个问题:
所以思来想去最保险的方法还是将业务镜像拉取下来,然后手动删除镜像中的 skywalking 包以及 JVM 参数,全部替换为 OpenTelemetry 的包和 JVM 参数。
整体的方案如下:
pod >0
的 deployment因为需要涉及到操作 kubernetes,所以整体就使用 Golang 实现了。
func ProcessDeployment(ctx context.Context, finish []string, deployment v1.Deployment, clientSet kubernetes.Interface) error {
deploymentName := deployment.Name
for _, s := range finish {
if s == deploymentName {
klog.Infof("Skip finish deployment:%s", deploymentName)
return nil
}
}
// Write finish deployment name to a file
defer writeDeploymentName2File(deploymentName, fmt.Sprintf("finish-%s.log", deployment.Namespace))
appName := deployment.GetObjectMeta().GetLabels()["appName"]
klog.Infof("Begin to process deployment:%s, appName:%s", deploymentName, appName)
upgrade, err := checkContainIstio(ctx, deployment, clientSet)
if err != nil {
return err
}
if upgrade == false {
klog.Infof("Don't have istio, No need to upgrade deployment:%s appName:%s", deploymentName, appName)
return nil
}
for i, container := range deployment.Spec.Template.Spec.Containers {
if strings.HasPrefix(deploymentName, container.Name) {
// Check if container has sw jvm
for _, envVar := range container.Env {
if envVar.Name == "CATALINA_OPTS" {
if !strings.Contains(envVar.Value, "skywalking") {
klog.Infof("Skip upgrade don't have sw jvm deployment:%s container:%s", deploymentName, container.Name)
return nil
}
}
}
upgrade(container)
// Check newDeployment status
go checkNewDeploymentStatus(ctx, clientSet, newDeployment)
// delete from image
deleteImage(container.Image)
}
}
return nil
}
这个函数需要传入一个 deployment ,同时还有一个已经完成了的列表进来。
已完成列表用于多次运行的时候可以快速跳过已经执行的 deployment。
checkContainIstio()
函数很简单,判断是否包含了 Istio 容器,如果没有包含说明不是后端应用(可能是前端、大数据之类的任务),就可以直接跳过了。
而判断是否需要替换的前提这事判断环境变量 CATALINA_OPTS
中是否包含了 skywalking 的内容,如果包含则说明需要进行替换。
func upgrade(container Container){
klog.Infof("Begin to upgrade deployment:%s container:%s", deploymentName, container.Name)
newImageName := fmt.Sprintf("%s-otel-%s", container.Image, generateRandomString(4))
err := BuildNewOtelImage(container.Image, newImageName)
if err != nil {
return err
}
// Update deployment jvm ENV
for e, envVar := range container.Env {
if envVar.Name == "CATALINA_OPTS" {
otelJVM := replaceSWAgent2OTel(envVar.Value, appName)
deployment.Spec.Template.Spec.Containers[i].Env[e].Value = otelJVM
}
}
// Update deployment image
deployment.Spec.Template.Spec.Containers[i].Image = newImageName
newDeployment, err := clientSet.AppsV1().Deployments(deployment.Namespace).Update(ctx, &deployment, metav1.UpdateOptions{})
if err != nil {
return err
}
klog.Infof("Finish upgrade deployment:%s container:%s", deploymentName, container.Name)
}
这里一共分为以下几部:
CATALINA_OPTS
环境变量,也就是替换 skywalking 的参数 dockerfile = fmt.Sprintf(`FROM %s
COPY %s /home/admin/%s
COPY otel.tar.gz /home/admin/otel.tar.gz
RUN tar -zxvf /home/admin/otel.tar.gz -C /home/admin
RUN rm -rf /home/admin/skywalking-agent
ENTRYPOINT ["/bin/sh", "/home/admin/start.sh"]
`, fromImage, script, script)
idx := strings.LastIndex(newImageName, "/") + 1
dockerFileName := newImageName[idx:]
create, err := os.Create(fmt.Sprintf("Dockerfile-%s", dockerFileName))
if err != nil {
return err
}
defer func() {
create.Close()
os.Remove(create.Name())
}()
_, err = create.WriteString(dockerfile)
if err != nil {
return err
}
cmd := exec.Command("docker", "build", ".", "-f", create.Name(), "-t", newImageName)
cmd.Stdin = strings.NewReader(dockerfile)
if err := cmd.Run(); err != nil {
return err
}
其实这里的重点就是构建这个新镜像,从这个 dockerfile 中也能看出具体的逻辑,也就是上文提到的删除原有的 skywalking 资源同时将新的 OpenTelemetry 资源打包进去。
最后再将这个镜像上传到私服。
其中的替换 JVM 参数也比较简单,直接删除 skywalking 的内容,然后再追加上 OpenTelemetry 需要的参数即可。
func checkNewDeploymentStatus(ctx context.Context, clientSet kubernetes.Interface, newDeployment *v1.Deployment) error {
ready := true
tick := time.Tick(10 * time.Second)
for i := 0; i < 30; i++ {
<-tick
originPodList, err := clientSet.CoreV1().Pods(newDeployment.Namespace).List(ctx, metav1.ListOptions{
LabelSelector: metav1.FormatLabelSelector(&metav1.LabelSelector{
MatchLabels: newDeployment.Spec.Selector.MatchLabels,
}),
})
if err != nil {
return err
}
// Check if there are any Pods
if len(originPodList.Items) == 0 {
klog.Infof("No Pod in deployment:%s, Skip", newDeployment.Name)
}
for _, item := range originPodList.Items {
// Check Pod running
for _, status := range item.Status.ContainerStatuses {
if status.RestartCount > 0 {
ready = false
break
}
}
}
klog.Infof("Check deployment:%s namespace:%s status:%t", newDeployment.Name, newDeployment.Namespace, ready)
if ready == false {
break
}
}
if ready == false {
// rollback
klog.Infof("=======Rollback deployment:%s namespace:%s", newDeployment.Name, newDeployment.Namespace)
writeDeploymentName2File(newDeployment.Name, fmt.Sprintf("rollback-%s.log", newDeployment.Namespace))
}
return nil
}
这里会启动一个 10s 执行一次的定时任务,每次都会检测是否有容器发生了重启(正常情况下是不会出现重启的)
如果检测了 30 次都没有重启的容器,那就说明本次替换成功了,不然就记录一个日志文件,然后人工处理。
这种通常是原有的镜像与 OpenTelemetry 不兼容,比如里面写死了一些 skywalking 的 API,导致启动失败。
所以替换任务跑完之后我还会检测这个 rollback-$namespace
的日志文件,人工处理这些失败的应用。
最后讲讲如何单个调用刚才的 ProcessDeployment()
函数。
考虑到不能对 kubernetes 产生影响,所以我们需要限制并发处理 deployment 的数量(我这里的限制是 10 个)。
所以就得分批进行替换,每次替换 10 个,而且其中有一个执行失败就得暂停后续任务,由人工检测失败原因再决定是否继续处理。
毕竟处理的是线上应用,需要小心谨慎。
所以触发的代码如下:
func ProcessDeploymentList(ctx context.Context, data []v1.Deployment, clientSet kubernetes.Interface) error {
file, err := os.ReadFile(fmt.Sprintf("finish-%s.log", data[0].Namespace))
if err != nil {
return err
}
split := strings.Split(string(file), "\n")
batchSize := 10
start := 0
for start < len(data) {
end := start + batchSize
if end > len(data) {
end = len(data)
}
batch := data[start:end]
//等待goroutine结束
var wg sync.WaitGroup
klog.Infof("Start process batch size %d", len(batch))
errs := make(chan error, len(batch))
wg.Add(len(batch))
for _, item := range batch {
d := item
go func() {
defer wg.Done()
if err := ProcessDeployment(ctx, split, d, clientSet); err != nil {
klog.Errorf("!!!Process deployment name:%s error: %v", d.Name, err)
errs <- err
return
}
}()
}
go func() {
wg.Wait()
close(errs)
}()
//任何一个失败就返回
for err := range errs {
if err != nil {
return err
}
}
start = end
klog.Infof("Deal next batch")
}
return nil
}
使用 WaitGroup
来控制一组任务,使用一个 chan 来传递异常;这类分批处理的代码在一些批处理框架中还蛮常见的。
最后只需要查询某个 namespace 下的所有 deployment 列表传入这个批处理函数即可。
不过整个过程中还是有几个点需要注意:
其实这个功能依然有提升空间,考虑到后续会升级 OpenTelemetry agent 的版本,甚至也需要增减一些 JVM 参数。
所以最后有一个统一的工具,可以直接升级 Agent,而不是每次我都需要修改这里的代码。
后来在网上看到了得物的相关分享,他们可以远程加载配置来解决这个问题。
这也是一种解决方案,直到我们看到了 OpenTelemetry 社区提供了 Operator,其中也包含了注入 agent 的功能。
apiVersion: opentelemetry.io/v1alpha1
kind: Instrumentation
metadata:
name: my-instrumentation
spec:
exporter:
endpoint: http://otel-collector:4317
propagators:
- tracecontext
- baggage
- b3
sampler:
type: parentbased_traceidratio
argument: "0.25"
java:
image: private/autoinstrumentation-java:1.32.0-1
我们可以使用他提供的 CRD 来配置我们 agent,只要维护好自己的镜像就好了。
使用起来也很简单,只要安装好了 OpenTelemetry-operator ,然后再需要注入 Java Agent 的 Pod 中使用注解:
instrumentation.opentelemetry.io/inject-java: "true"
operator 就会自动从刚才我们配置的镜像中读取 agent,然后复制到我们的业务容器。
再配置上环境变量 $JAVA_TOOL_OPTIONS=/otel/javaagent.java
, 这是一个 Java 内置的环境变量,应用启动的时候会自动识别,这样就可以自动注入 agent 了。
envJavaToolsOptions = "JAVA_TOOL_OPTIONS"
// set env value
idx := getIndexOfEnv(container.Env, envJavaToolsOptions)
if idx == -1 {
container.Env = append(container.Env, corev1.EnvVar{
Name: envJavaToolsOptions,
Value: javaJVMArgument,
})} else {
container.Env[idx].Value = container.Env[idx].Value + javaJVMArgument
}
// copy javaagent.jar
pod.Spec.InitContainers = append(pod.Spec.InitContainers, corev1.Container{
Name: javaInitContainerName,
Image: javaSpec.Image,
Command: []string{"cp", "/javaagent.jar", javaInstrMountPath + "/javaagent.jar"},
Resources: javaSpec.Resources,
VolumeMounts: []corev1.VolumeMount{{
Name: javaVolumeName,
MountPath: javaInstrMountPath,
}},})
大致的运行原理是当有 Pod 的事件发生了变化(重启、重新部署等),operator 就会检测到变化,此时会判断是否开启了刚才的注解:
instrumentation.opentelemetry.io/inject-java: "true"
接着会写入环境变量 JAVA_TOOL_OPTIONS
,同时将 jar 包从 InitContainers 中复制到业务容器中。
这里使用到了 kubernetes 的初始化容器,该容器是用于做一些准备工作的,比如依赖安装、配置检测或者是等待其他一些组件启动成功后再启动业务容器。
目前这个 operator 还处于试用阶段,同时部分功能还不满足(比如支持自定义扩展),今后有时间也可以分析下它的运行原理。
参考链接:
本文分享自 crossoverJie 微信公众号,前往查看
如有侵权,请联系 cloudcommunity@tencent.com 删除。
本文参与 腾讯云自媒体同步曝光计划 ,欢迎热爱写作的你一起参与!
扫码关注腾讯云开发者
领取腾讯云代金券
Copyright © 2013 - 2025 Tencent Cloud. All Rights Reserved. 腾讯云 版权所有
深圳市腾讯计算机系统有限公司 ICP备案/许可证号:粤B2-20090059 深公网安备号 44030502008569
腾讯云计算(北京)有限责任公司 京ICP证150476号 | 京ICP备11018762号 | 京公网安备号11010802020287
Copyright © 2013 - 2025 Tencent Cloud.
All Rights Reserved. 腾讯云 版权所有