- VisualStudio2022插件的安装及使用-编程手把手系列文章
- pprof-在现网场景怎么用
- C#实现的下拉多选框,下拉多选树,多级节点
- 【学习笔记】基础数据结构:猫树
最近公司将我们之前使用的链路工具切换为了 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 还处于使用阶段,同时部分功能还不满足(比如支持自定义扩展),今后有时间也可以分析下它的运行原理.
参考链接:
最后此篇关于实战:如何优雅的从Skywalking切换到OpenTelemetry的文章就讲到这里了,如果你想了解更多关于实战:如何优雅的从Skywalking切换到OpenTelemetry的内容请搜索CFSDN的文章或继续浏览相关文章,希望大家以后支持我的博客! 。
实战-行业攻防应急响应 简介: 服务器场景操作系统 Ubuntu 服务器账号密码:root/security123 分析流量包在/home/security/security.pcap 相
背景 最近公司将我们之前使用的链路工具切换为了 OpenTelemetry. 我们的技术栈是: OTLP C
一 同一类的方法都用 synchronized 修饰 1 代码 package concurrent; import java.util.concurrent.TimeUnit; public c
一 简单例子 1 代码 package concurrent.threadlocal; /** * ThreadLocal测试 * * @author cakin */ public class T
1. 问题背景 问题发生在快递分拣的流程中,我尽可能将业务背景简化,让大家只关注并发问题本身。 分拣业务针对每个快递包裹都会生成一个任务,我们称它为 task。task 中有两个字段需要
实战环境 elastic search 8.5.0 + kibna 8.5.0 + springboot 3.0.2 + spring data elasticsearch 5.0.2 +
Win10下yolov8 tensorrt模型加速部署【实战】 TensorRT-Alpha 基于tensorrt+cuda c++实现模型end2end的gpu加速,支持win10、
yolov8 tensorrt模型加速部署【实战】 TensorRT-Alpha 基于tensorrt+cuda c++实现模型end2end的gpu加速,支持win10、linux,
目录如下: 为什么需要自定义授权类型? 前面介绍OAuth2.0的基础知识点时介绍过支持的4种授权类型,分别如下: 授权码模式 简化模式 客户端模式 密码模式
今天这篇文章介绍一下如何在修改密码、修改权限、注销等场景下使JWT失效。 文章的目录如下: 解决方案 JWT最大的一个优势在于它是无状态的,自身包含了认证鉴权所需要的所有信息,服务器端
前言 大家好,我是捡田螺的小男孩。(求个星标置顶) 我们日常做分页需求时,一般会用limit实现,但是当偏移量特别大的时候,查询效率就变得低下。本文将分四个方案,讨论如何优化MySQL百万数
前言 大家好,我是捡田螺的小男孩。 平时我们写代码呢,多数情况都是流水线式写代码,基本就可以实现业务逻辑了。如何在写代码中找到乐趣呢,我觉得,最好的方式就是:使用设计模式优化自己
我们先讲一些arm汇编的基础知识。(我们以armv7为例,最新iphone5s上的64位暂不讨论) 基础知识部分: 首先你介绍一下寄存器: r0-r3:用于函数参数及返回值的传递 r4-r6
一 同一类的静态方法都用 synchronized 修饰 1 代码 package concurrent; import java.util.concurrent.TimeUnit; public
DRF快速写五个接口,比你用手也快··· 实战-DRF快速写接口 开发环境 Python3.6 Pycharm专业版2021.2.3 Sqlite3 Django 2.2 djangorestfram
一 添加依赖 org.apache.thrift libthrift 0.11.0 二 编写 IDL 通过 IDL(.thrift 文件)定义数据结构、异常和接口等数据,供各种编程语言使用 nam
我正在阅读 Redis in action e-book关于semaphores的章节.这是使用redis实现信号量的python代码 def acquire_semaphore(conn, semn
自定义控件在WPF开发中是很常见的,有时候某些控件需要契合业务或者美化统一样式,这时候就需要对控件做出一些改造。 目录 按钮设置圆角
师父布置的任务,让我写一个服务练练手,搞清楚socket的原理和过程后跑了一个小demo,很有成就感,代码内容也比较清晰易懂,很有教育启发意义。 代码 ?
? 1 2
我是一名优秀的程序员,十分优秀!