DeltaFIFO makes MemoryLeak

[huhu-sky]

What i found

i import java client sdk and use list-watch function with informer; while my application running for about several hours, it occured gc all the time; I dump the hprof file to the EclipseMAT tool and found some memory leak suspects.

How can i resolve this problem, or is there something i wrong; Thanks .

public class ListWatchInformer {

    final K8sService k8s;
    SharedInformerFactory svcFactory;
    SharedInformerFactory epFactory;
    SharedInformerFactory nacosSvcFactory;

    private SharedIndexInformer<V1Service> serviceInformer;
    private SharedIndexInformer<V1Endpoints> endpointsInformer;
    private SharedIndexInformer<V1Service> nacosServiceInformer;

    @Autowired
    public ListWatchInformer(K8sService k8s) {
        this.k8s = k8s;
        init();
    }

    public void init() {
        this.svcFactory = new SharedInformerFactory();
        this.epFactory = new SharedInformerFactory();
        this.nacosSvcFactory = new SharedInformerFactory();
        CoreV1Api coreV1Api = k8s.getConn(ServiceConstants.ClusterId_Test).coreV1Api;
        // service list watch
        this.serviceInformer = svcInformer(svcFactory, coreV1Api, ServiceConstants.FixedNacosK8sLabelSelector);
        svcFactory.startAllRegisteredInformers();
        // endpoints list watch
        this.endpointsInformer = epInformer(epFactory, coreV1Api, ServiceConstants.FixedNacosK8sLabelSelector);
        epFactory.startAllRegisteredInformers();
        // nacos service list watch
        this.nacosServiceInformer = namespacedServiceInformer(nacosSvcFactory, ServiceConstants.NacosNamespace, coreV1Api, null);
        nacosSvcFactory.startAllRegisteredInformers();
    }


    public SharedIndexInformer<V1Service> svcInformer() {
        return serviceInformer;
    }

    public SharedIndexInformer<V1Endpoints> epInformer() {
        return endpointsInformer;
    }

    public SharedIndexInformer<V1Service> nacosSvcInformer() {
        return nacosServiceInformer;
    }

    public SharedInformerFactory epFactory(){
        return epFactory;
    }
    public SharedInformerFactory svcFactory(){
        return svcFactory;
    }
    public SharedInformerFactory nacosSvcFactory(){
        return nacosSvcFactory;
    }

    public V1Service getSvc(String namespace, String name) {
        String key = namespace + "/" + name;
        return getSvc(key);
    }

    public V1Service getNacosSvc(String name) {
        if (nacosServiceInformer != null && nacosServiceInformer.getIndexer() != null) {
            return nacosServiceInformer.getIndexer().getByKey(ServiceConstants.NacosNamespace + "/" + name);
        }
        return null;
    }

    public V1Service getSvc(String key) {
        if (serviceInformer != null && serviceInformer.getIndexer() != null) {
            return serviceInformer.getIndexer().getByKey(key);
        }
        return null;
    }

    public V1Endpoints getEp(String namespace, String name) {
        String key = namespace + "/" + name;
        return getEp(key);
    }

    public V1Endpoints getEp(String key) {
        if (endpointsInformer != null && endpointsInformer.getIndexer() != null) {
            return endpointsInformer.getIndexer().getByKey(key);
        }
        return null;
    }

    private SharedIndexInformer<V1Endpoints> epInformer(SharedInformerFactory factory, CoreV1Api coreV1Api, String label) {
        // service informer
        SharedIndexInformer<V1Endpoints> endpointsInformer =
                factory.sharedIndexInformerFor(
                        (CallGeneratorParams params) -> {
                            try {
                                return coreV1Api.listEndpointsForAllNamespacesCall(null, null, null, label,
                                        null, null, params.resourceVersion, params.timeoutSeconds, params.watch, null,null);
                            } catch (ApiException e) {
                                Loggers.K8S_LOG.error("endpointsInformer: endpoints list error: {}", e);
                                Loggers.K8S_LOG.error("endpointsInformer: endpoints list error, responseBody: {}, message: {}, stack: {}", e.getResponseBody(), e.getMessage(),e.getStackTrace());
                            }
                            return null;
                        },
                        V1Endpoints.class,
                        V1EndpointsList.class);

        endpointsInformer.addEventHandler(epHandler());
        return endpointsInformer;
    }

    private SharedIndexInformer<V1Service> svcInformer(SharedInformerFactory factory, CoreV1Api coreV1Api, String label) {
        // service informer
        SharedIndexInformer<V1Service> serviceInformer =
                factory.sharedIndexInformerFor(
                        (CallGeneratorParams params) -> {
                            try {
                                return coreV1Api.listServiceForAllNamespacesCall(null, null, null, label,
                                        null, null, params.resourceVersion, params.timeoutSeconds, params.watch, null,null);
                            } catch (ApiException e) {
                                Loggers.K8S_LOG.error("serviceInformer: service list error: {}", e);
                                Loggers.K8S_LOG.error("serviceInformer: service list error, responseBody: {}, message: {}, stack: {}", e.getResponseBody(), e.getMessage(),e.getStackTrace());
                            }
                            return null;
                        },
                        V1Service.class,
                        V1ServiceList.class);

        serviceInformer.addEventHandler(serviceHandler());
        return serviceInformer;
    }

    private SharedIndexInformer<V1Service> namespacedServiceInformer(SharedInformerFactory factory, String namespace, CoreV1Api coreV1Api, String label) {
        // service informer
        SharedIndexInformer<V1Service> serviceInformer =
                factory.sharedIndexInformerFor(
                        (CallGeneratorParams params) -> {
                            try {
                                return coreV1Api.listNamespacedServiceCall(namespace, null, null, null, null,
                                        label, null, params.resourceVersion, params.timeoutSeconds, params.watch, null,null);
                            } catch (ApiException e) {
                                Loggers.K8S_LOG.error("serviceInformer: service list error: {}", e);
                                Loggers.K8S_LOG.error("serviceInformer: service list error, responseBody: {}, message: {}, stack: {}", e.getResponseBody(), e.getMessage(),e.getStackTrace());
                            }
                            return null;
                        },
                        V1Service.class,
                        V1ServiceList.class);

        serviceInformer.addEventHandler(serviceHandler());
        return serviceInformer;
    }

    private ResourceEventHandler<V1Endpoints> epHandler() {
        return new ResourceEventHandler<V1Endpoints>() {
            @Override
            public void onAdd(V1Endpoints ep) {
                if (Loggers.K8S_LOG.isInfoEnabled()) {
                    Loggers.K8S_LOG.info("endpointsInformer: ep added!, namespace: {}, name: {}, origin: {}",
                            ep.getMetadata().getNamespace(), ep.getMetadata().getName(), ep.getSubsets());
                }
            }

            @Override
            public void onUpdate(V1Endpoints ep, V1Endpoints newEp) {
                if (Loggers.K8S_LOG.isInfoEnabled()) {
                    if (!ep.getMetadata().getResourceVersion().equals(newEp.getMetadata().getResourceVersion())) {
                        Loggers.K8S_LOG.info("endpointsInformer: ep updated!, namespace: {}, name: {}, oldSubsets: {}, newSubsets: {}, oldVersion: {}, newVersion: {}",
                                ep.getMetadata().getNamespace(), ep.getMetadata().getName(), ep.getSubsets(), newEp.getSubsets(),
                                ep.getMetadata().getResourceVersion(), newEp.getMetadata().getResourceVersion());
                    }
                }
            }

            @Override
            public void onDelete(V1Endpoints ep, boolean deletedFinalStateUnknown) {
                if (Loggers.K8S_LOG.isInfoEnabled()) {
                    Loggers.K8S_LOG.info("endpointsInformer: ep deleted!, namespace: {}, name: {}, origin: {}",
                            ep.getMetadata().getNamespace(), ep.getMetadata().getName(), ep.getSubsets());
                }
            }
        };
    }

    private ResourceEventHandler<V1Service> serviceHandler() {
        return new ResourceEventHandler<V1Service>() {
            @Override
            public void onAdd(V1Service srv) {
                if (Loggers.K8S_LOG.isInfoEnabled()) {
                    Loggers.K8S_LOG.info("serviceInformer: srv added!, namespace: {}, name: {}, origin: {}",
                            srv.getMetadata().getNamespace(), srv.getMetadata().getName(),
                            srv.getMetadata().getAnnotations().get(ServiceConstants.Nacos2K8sServiceNameAnno_Key));
                }
            }

            @Override
            public void onUpdate(V1Service srv, V1Service newSrv) {
                if (Loggers.K8S_LOG.isInfoEnabled()) {
                    if (!srv.getMetadata().getResourceVersion().equals(newSrv.getMetadata().getResourceVersion())) {
                        Loggers.K8S_LOG.info("serviceInformer: srv updated!, namespace: {}, name: {}, origin: {}, oldVersion: {}, newVersion: {}",
                                srv.getMetadata().getNamespace(), srv.getMetadata().getName(),
                                srv.getMetadata().getAnnotations().get(ServiceConstants.Nacos2K8sServiceNameAnno_Key),
                                srv.getMetadata().getResourceVersion(), newSrv.getMetadata().getResourceVersion());
                        if (Loggers.K8S_LOG.isDebugEnabled()) {
                            Loggers.K8S_LOG.info("serviceInformer: srv updated!, namespace: {}, name: {}, origin: {}, old: {}, new: {}",
                                    srv.getMetadata().getNamespace(), srv.getMetadata().getName(),
                                    srv.getMetadata().getAnnotations().get(ServiceConstants.Nacos2K8sServiceNameAnno_Key),
                                    srv, newSrv);
                        }
                    }
                }
            }

            @Override
            public void onDelete(V1Service srv, boolean deletedFinalStateUnknown) {
                if (Loggers.K8S_LOG.isInfoEnabled()) {
                    Loggers.K8S_LOG.info("serviceInformer: srv deleted!, namespace: {}, name: {}, origin: {}",
                            srv.getMetadata().getNamespace(), srv.getMetadata().getName(),
                            srv.getMetadata().getAnnotations().get(ServiceConstants.Nacos2K8sServiceNameAnno_Key));
                }
            }
        };
    }
}

[huhu-sky]

the client version is 5.0

[brendandburns]

cc @yue9944882

[yue9944882]

I dump the hprof file to the EclipseMAT tool

thank you for providing the detailed dump. as far as i can see from it, there's a few (i can't see the actual number of objects from the figure, sry) objects references in DeltaFIFO#items, which possible means that the informer is too busy to catch/sync up w/ the latest events from the watch connection. can you verify if there's exactly the same number of object references kept in DeltaFIFO#queue?

i import java client sdk and use list-watch function with informer; while my application running for about several hours, it occured gc all the time; I dump the hprof file to the EclipseMAT tool and found some memory leak suspects.

i don't think there's actual memory leaks as we've been running this in production for years. but frequent GC can be a problem sometimes, b/c every-time the list-watcher receives a notification from the watch connection, it instantiates a new object and deserializes into it. the number of dangling objects (pending for GC) can be rather bigger in some extra busy clusters..

one way to ease the GC pressure is to develop a custom Gson#ExclusionStrategy so that we only deserialize a few fields we cares.

[huhu-sky]

i don't think there's actual memory leaks as we've been running this in production for years.

thanks for your replying;
first thing is that I found i can receive service update event even the old service version is equals with new service version;

I cant understand about it;

[huhu-sky]

i also found that there about 596 services which are the same service with the same version, so i think if the -Xmx and old generation is more bigger, the sample number of the same service be more than this sample.

[fejta-bot]

Issues go stale after 90d of inactivity.
Mark the issue as fresh with /remove-lifecycle stale.
Stale issues rot after an additional 30d of inactivity and eventually close.

If this issue is safe to close now please do so with /close.

Send feedback to sig-testing, kubernetes/test-infra and/or fejta.
/lifecycle stale

[fejta-bot]

Stale issues rot after 30d of inactivity.
Mark the issue as fresh with /remove-lifecycle rotten.
Rotten issues close after an additional 30d of inactivity.

If this issue is safe to close now please do so with /close.

Send feedback to sig-testing, kubernetes/test-infra and/or fejta.
/lifecycle rotten

[fishautumn]

I met this issue recently. the reason is each time connection timeout, it reconnects and calls list method, then a new copy is added to every cached objects. memory won't get freed until object deleted from k8s.

this is my proposed fix: #1259